diff --git a/cmd/drive9-server/main.go b/cmd/drive9-server/main.go index e556537f..586cbec1 100644 --- a/cmd/drive9-server/main.go +++ b/cmd/drive9-server/main.go @@ -43,8 +43,10 @@ import ( ) const ( - defaultListenAddr = ":9009" - defaultS3Dir = "s3" + defaultListenAddr = ":9009" + defaultS3Dir = "s3" + defaultTenantBackendCacheMaxTenants = 1024 + defaultSharedBackendCacheMaxTenants = 20480 ) type s3Config struct { @@ -341,7 +343,7 @@ func main() { S3SessionToken: s3cfg.SessionToken, S3EncryptionPolicy: s3cfg.EncryptionPolicy, BackendOptions: backendOptions, - MaxTenants: envInt("DRIVE9_POOL_MAX_TENANTS", 0), + MaxTenants: tenantBackendCacheMaxTenants(providerType), IdleTimeout: envDuration("DRIVE9_POOL_IDLE_TTL", 5*time.Minute), IdleReapInterval: envDuration("DRIVE9_POOL_IDLE_REAP_INTERVAL", 2*time.Minute), DisableDatabaseAutoEmbedding: disableDatabaseAutoEmbedding, @@ -621,8 +623,8 @@ environment: DRIVE9_LISTEN_ADDR serve listen address (default: :9009) DRIVE9_PUBLIC_URL externally reachable base URL for presigned URLs (required for remote clients) DRIVE9_META_DSN control-plane MySQL DSN (required) - DRIVE9_POOL_MAX_TENANTS max cached tenant user DB pools per pod (default: 1024) - DRIVE9_POOL_IDLE_TTL idle duration before a cached tenant backend is evicted (default: 5m, 0=disabled) + DRIVE9_POOL_MAX_TENANTS max cached tenant backends per pod (default: 20480 for tidb_cloud_native_shared, 1024 otherwise) + DRIVE9_POOL_IDLE_TTL idle duration before a standalone tenant backend is evicted (default: 5m, 0=disabled; shared tenants use capacity LRU only) DRIVE9_POOL_IDLE_REAP_INTERVAL how often the idle reaper scans (default: 2m) DRIVE9_META_DB_MAX_OPEN_CONNS max open connections for the per-pod meta DB pool (default: 100) DRIVE9_META_DB_MAX_IDLE_CONNS max idle connections for the per-pod meta DB pool (default: 20) @@ -1156,6 +1158,18 @@ func dbHealthProbeOptionsFromEnv() metrics.DBHealthProbeOptions { } } +func tenantBackendCacheMaxTenants(provider string) int { + fallback := defaultTenantBackendCacheMaxTenants + if provider == tenant.ProviderTiDBCloudNativeShared { + fallback = defaultSharedBackendCacheMaxTenants + } + maxTenants := envInt("DRIVE9_POOL_MAX_TENANTS", fallback) + if maxTenants <= 0 { + return fallback + } + return maxTenants +} + func envInt(key string, fallback int) int { raw := strings.TrimSpace(os.Getenv(key)) if raw == "" { diff --git a/cmd/drive9-server/main_test.go b/cmd/drive9-server/main_test.go index bcade259..62432fd4 100644 --- a/cmd/drive9-server/main_test.go +++ b/cmd/drive9-server/main_test.go @@ -9,6 +9,7 @@ import ( "github.com/mem9-ai/drive9/pkg/backend" "github.com/mem9-ai/drive9/pkg/meta" "github.com/mem9-ai/drive9/pkg/server" + "github.com/mem9-ai/drive9/pkg/tenant" ) func TestVersionTextUsesDrive9ServerComponent(t *testing.T) { @@ -18,6 +19,82 @@ func TestVersionTextUsesDrive9ServerComponent(t *testing.T) { } } +func TestTenantBackendCacheMaxTenantsUsesProviderDefault(t *testing.T) { + const key = "DRIVE9_POOL_MAX_TENANTS" + restore := snapshotEnv(t, []string{key}) + t.Cleanup(func() { restoreEnv(t, restore) }) + unsetEnv(t, []string{key}) + + tests := []struct { + provider string + want int + }{ + {provider: tenant.ProviderTiDBCloudNativeShared, want: 20480}, + {provider: tenant.ProviderTiDBCloudNative, want: 1024}, + {provider: tenant.ProviderTiDBZero, want: 1024}, + {provider: tenant.ProviderDB9, want: 1024}, + } + for _, tt := range tests { + t.Run(tt.provider, func(t *testing.T) { + if got := tenantBackendCacheMaxTenants(tt.provider); got != tt.want { + t.Errorf("tenantBackendCacheMaxTenants(%q) = %d, want %d", tt.provider, got, tt.want) + } + }) + } +} + +func TestTenantBackendCacheMaxTenantsExplicitOverrideWins(t *testing.T) { + const key = "DRIVE9_POOL_MAX_TENANTS" + restore := snapshotEnv(t, []string{key}) + t.Cleanup(func() { restoreEnv(t, restore) }) + setEnv(t, key, "4096") + + for _, provider := range []string{ + tenant.ProviderTiDBCloudNativeShared, + tenant.ProviderTiDBCloudNative, + tenant.ProviderTiDBZero, + tenant.ProviderDB9, + } { + if got := tenantBackendCacheMaxTenants(provider); got != 4096 { + t.Errorf("tenantBackendCacheMaxTenants(%q) = %d, want explicit 4096", provider, got) + } + } +} + +func TestTenantBackendCacheMaxTenantsFallsBackForInvalidOverrides(t *testing.T) { + const key = "DRIVE9_POOL_MAX_TENANTS" + restore := snapshotEnv(t, []string{key}) + t.Cleanup(func() { restoreEnv(t, restore) }) + + for _, raw := range []string{"0", "-1", "bad"} { + setEnv(t, key, raw) + if got := tenantBackendCacheMaxTenants(tenant.ProviderTiDBCloudNativeShared); got != 20480 { + t.Errorf("tenantBackendCacheMaxTenants(shared) with %q = %d, want 20480", raw, got) + } + if got := tenantBackendCacheMaxTenants(tenant.ProviderTiDBCloudNative); got != 1024 { + t.Errorf("tenantBackendCacheMaxTenants(native) with %q = %d, want 1024", raw, got) + } + } + + unsetEnv(t, []string{key}) + if got := tenantBackendCacheMaxTenants("unknown-provider"); got != 1024 { + t.Errorf("tenantBackendCacheMaxTenants(unknown-provider) = %d, want 1024", got) + } +} + +func TestRestoreEnvPreservesPresentEmptyValue(t *testing.T) { + const key = "DRIVE9_TEST_PRESENT_EMPTY_ENV" + t.Setenv(key, "") + snapshot := snapshotEnv(t, []string{key}) + unsetEnv(t, []string{key}) + + restoreEnv(t, snapshot) + value, present := os.LookupEnv(key) + if !present || value != "" { + t.Errorf("restored env = (%q, %t), want present empty value", value, present) + } +} + func TestSlockOAuthFromEnvDisabledByDefault(t *testing.T) { keys := []string{ "DRIVE9_SLOCK_ORIGIN", @@ -682,25 +759,31 @@ func TestS3ConfigValidateRejectsInvalidStaticCredentialCombinations(t *testing.T } } -func snapshotEnv(t *testing.T, keys []string) map[string]string { +type envSnapshotValue struct { + value string + present bool +} + +func snapshotEnv(t *testing.T, keys []string) map[string]envSnapshotValue { t.Helper() - out := make(map[string]string, len(keys)) + out := make(map[string]envSnapshotValue, len(keys)) for _, key := range keys { - out[key] = os.Getenv(key) + value, present := os.LookupEnv(key) + out[key] = envSnapshotValue{value: value, present: present} } return out } -func restoreEnv(t *testing.T, snapshot map[string]string) { +func restoreEnv(t *testing.T, snapshot map[string]envSnapshotValue) { t.Helper() - for key, value := range snapshot { - if value == "" { + for key, entry := range snapshot { + if !entry.present { if err := os.Unsetenv(key); err != nil { t.Fatalf("unset %s: %v", key, err) } continue } - if err := os.Setenv(key, value); err != nil { + if err := os.Setenv(key, entry.value); err != nil { t.Fatalf("restore %s: %v", key, err) } } diff --git a/pkg/backend/llm_usage_test.go b/pkg/backend/llm_usage_test.go index 60ba1eda..0a9a49e8 100644 --- a/pkg/backend/llm_usage_test.go +++ b/pkg/backend/llm_usage_test.go @@ -28,10 +28,6 @@ func (m *mockQuotaStore) GetQuotaConfig(_ context.Context, _ string) (*QuotaConf return &QuotaConfigView{}, nil } -func (m *mockQuotaStore) GetQuotaConfigVersion(_ context.Context, _ string) (string, error) { - return "", nil -} - func (m *mockQuotaStore) GetQuotaUsage(_ context.Context, _ string) (*QuotaUsageView, error) { return &QuotaUsageView{}, nil } diff --git a/pkg/backend/options.go b/pkg/backend/options.go index 67e0a8af..73344bf9 100644 --- a/pkg/backend/options.go +++ b/pkg/backend/options.go @@ -21,10 +21,10 @@ const ( defaultMaxExtractedTextBytes = DefaultImageExtractMaxTextBytes defaultAudioExtractMaxSize = int64(32 << 20) // 32 MiB defaultAudioExtractTimeout = 2 * time.Minute - defaultMaxAudioExtractedTextBytes = 8 << 10 // 8 KiB + defaultMaxAudioExtractedTextBytes = 8 << 10 // 8 KiB defaultVideoExtractMaxSize = int64(200 << 20) // 200 MiB defaultVideoExtractTimeout = 5 * time.Minute - defaultMaxVideoExtractedTextBytes = 32 << 10 // 32 KiB + defaultMaxVideoExtractedTextBytes = 32 << 10 // 32 KiB defaultMaxUploadBytes = int64(10 * (1 << 30)) // 10 GiB defaultMaxTenantStorageBytes = int64(50 * (1 << 30)) // 50 GiB defaultMaxMediaLLMFiles = int64(500) // 500 media files per tenant @@ -393,7 +393,6 @@ func (b *Dat9Backend) Close() { } b.stopMutationWorker() if b.quotaConfigCache != nil { - b.quotaConfigCache.stop() b.quotaConfigCache = nil } } diff --git a/pkg/backend/quota.go b/pkg/backend/quota.go index e90613c8..735e4c9a 100644 --- a/pkg/backend/quota.go +++ b/pkg/backend/quota.go @@ -228,10 +228,7 @@ func (b *Dat9Backend) checkStorageQuotaServerTx(ctx context.Context, tx *sql.Tx, // falling back to a synchronous DB query when the cache is unavailable. func (b *Dat9Backend) cachedQuotaConfig(ctx context.Context) *QuotaConfigView { if b.quotaConfigCache != nil { - if cfg := b.quotaConfigCache.get(); cfg != nil { - return cfg - } - return b.quotaConfigCache.load(ctx) + return b.quotaConfigCache.get(ctx) } if b.metaStore == nil { return nil diff --git a/pkg/backend/quota_cache.go b/pkg/backend/quota_cache.go index 3bd961b3..d6d25d26 100644 --- a/pkg/backend/quota_cache.go +++ b/pkg/backend/quota_cache.go @@ -2,6 +2,7 @@ package backend import ( "context" + "math/rand/v2" "sync" "time" @@ -12,10 +13,24 @@ import ( ) const ( - // defaultQuotaConfigCacheRefreshInterval is the default interval for the - // background goroutine that polls the tenant quota config version from - // the central DB. Override with DRIVE9_QUOTA_CACHE_REFRESH_SECONDS. + // defaultQuotaConfigCacheRefreshInterval is the default TTL for lazily + // loaded tenant quota config. Override with DRIVE9_QUOTA_CACHE_REFRESH_SECONDS. defaultQuotaConfigCacheRefreshInterval = 30 * time.Second + // quotaConfigCacheLoadTimeout bounds a coalesced refresh independently of + // the request that happened to claim load ownership. + quotaConfigCacheLoadTimeout = 5 * time.Second + // defaultQuotaConfigCacheAsyncRefreshSlots bounds detached warm-cache + // refreshes so a MetaDB slowdown cannot create one in-flight query per + // resident tenant. Keep headroom in the shared metadata connection pool for + // foreground requests and other control-plane work. + defaultQuotaConfigCacheAsyncRefreshSlots = 32 + // quotaConfigCacheSlotRetryInterval prevents a tenant that could not claim + // an async refresh slot from retrying on every request while the budget is + // exhausted. + quotaConfigCacheSlotRetryInterval = time.Second + // quotaConfigCacheFailureRetryInterval keeps quota changes responsive after + // a transient MetaDB failure. Successful refreshes retain the normal TTL. + quotaConfigCacheFailureRetryInterval = 5 * time.Second // quotaUsageCacheTTL bounds how long soft small-write quota checks may // reuse central usage counters. Strict upload reservations still read // central usage directly. @@ -30,9 +45,10 @@ const ( // quotaConfigCacheRefreshInterval is the resolved refresh interval (package-level // var so it can be set from env at startup). var ( - quotaConfigCacheRefreshInterval = defaultQuotaConfigCacheRefreshInterval - quotaUsageCacheTTL = defaultQuotaUsageCacheTTL - quotaPendingDeltasCacheTTL = defaultQuotaPendingDeltasCacheTTL + quotaConfigCacheRefreshInterval = defaultQuotaConfigCacheRefreshInterval + quotaUsageCacheTTL = defaultQuotaUsageCacheTTL + quotaPendingDeltasCacheTTL = defaultQuotaPendingDeltasCacheTTL + quotaConfigCacheAsyncRefreshSlots = make(chan struct{}, defaultQuotaConfigCacheAsyncRefreshSlots) ) // InitQuotaConfigCacheRefreshInterval overrides the default refresh interval. @@ -60,11 +76,6 @@ func InitQuotaAdmissionCacheTTLs(usageTTL, pendingDeltasTTL time.Duration) { } } -type quotaConfigSnapshot struct { - config *QuotaConfigView - version string -} - func cloneQuotaConfigView(cfg *QuotaConfigView) *QuotaConfigView { if cfg == nil { return nil @@ -73,134 +84,219 @@ func cloneQuotaConfigView(cfg *QuotaConfigView) *QuotaConfigView { return &cp } -// quotaConfigCache is a per-tenant cache for low-frequency quota config. It -// only removes repeated config reads and uses version polling so config changes -// converge without a cross-server invalidation channel. +// quotaConfigCacheRefreshDelay spreads refreshes over the last 10% of the +// configured TTL. The delay never exceeds the configured value, preserving +// the maximum quota-config convergence window. +func quotaConfigCacheRefreshDelay(ttl time.Duration) time.Duration { + if ttl <= 0 { + return 0 + } + maxJitter := ttl / 10 + if maxJitter <= 0 { + return ttl + } + return ttl - time.Duration(rand.Int64N(int64(maxJitter)+1)) +} + +// quotaConfigCacheFailureRetryDelay spreads failed loads symmetrically around +// the five-second retry target. Unlike successful refreshes, failures may be +// retried slightly after the target so independently warmed tenants do not +// converge into a synchronized retry wave. +func quotaConfigCacheFailureRetryDelay(ttl time.Duration) time.Duration { + if ttl <= 0 { + return 0 + } + jitter := ttl / 10 + if jitter <= 0 { + return ttl + } + return ttl - jitter + time.Duration(rand.Int64N(int64(2*jitter)+1)) +} + +// quotaConfigCache is a passive per-tenant cache for low-frequency quota +// config. Requests refresh an expired snapshot; idle tenants create no +// goroutines and issue no quota queries. type quotaConfigCache struct { tenantID string tidbCloudOrgID string store MetaQuotaStore - mu sync.RWMutex - snapshot *quotaConfigSnapshot - loadMu sync.Mutex - - cancel context.CancelFunc - done chan struct{} + mu sync.RWMutex + snapshot *QuotaConfigView + nextRefresh time.Time + loadDone chan struct{} } -// newQuotaConfigCache creates and starts a background-refreshing config cache. -// Backend construction must stay cheap for read-only operations such as ls, so -// the initial load is lazy: the first quota check loads config on demand and -// the background refresher keeps it current after that. +// newQuotaConfigCache creates an empty request-driven config cache. Backend +// construction stays cheap: the first quota check loads config on demand. func newQuotaConfigCache(tenantID, tidbCloudOrgID string, store MetaQuotaStore) *quotaConfigCache { - ctx, cancel := context.WithCancel(context.Background()) - c := "aConfigCache{ + return "aConfigCache{ tenantID: tenantID, tidbCloudOrgID: normalizeTenantMetricTiDBCloudOrgID(tidbCloudOrgID), store: store, - cancel: cancel, - done: make(chan struct{}), } - go c.run(ctx) - return c } -// get returns a copy of the cached quota config. Returns nil when the cache has -// not been populated yet, allowing callers to fail open or fall back. -func (c *quotaConfigCache) get() *QuotaConfigView { +// cached returns the current snapshot and whether another store read is +// suppressed until nextRefresh. A nil snapshot can still be current during +// the short retry cooldown after an initial load failure. +func (c *quotaConfigCache) cached(now time.Time) (*QuotaConfigView, bool) { c.mu.RLock() defer c.mu.RUnlock() - if c.snapshot == nil || c.snapshot.config == nil { - return nil + if !now.Before(c.nextRefresh) { + return nil, false + } + if c.snapshot == nil { + return nil, true } - return cloneQuotaConfigView(c.snapshot.config) + return cloneQuotaConfigView(c.snapshot), true } -func (c *quotaConfigCache) load(ctx context.Context) *QuotaConfigView { - if cfg := c.get(); cfg != nil { +// get returns a defensive copy of the cached config, refreshing it once when +// its TTL has expired. Warm callers serve stale config while a refresh is in +// flight; cold callers wait only until their own context is done. +func (c *quotaConfigCache) get(ctx context.Context) *QuotaConfigView { + now := time.Now() + if cfg, current := c.cached(now); current { return cfg } - c.loadMu.Lock() - defer c.loadMu.Unlock() - if cfg := c.get(); cfg != nil { + + c.mu.Lock() + now = time.Now() + if now.Before(c.nextRefresh) { + cfg := cloneQuotaConfigView(c.snapshot) + c.mu.Unlock() return cfg } + if c.loadDone != nil { + done := c.loadDone + stale := cloneQuotaConfigView(c.snapshot) + c.mu.Unlock() + if stale != nil { + return stale + } + select { + case <-done: + c.mu.RLock() + cfg := cloneQuotaConfigView(c.snapshot) + c.mu.RUnlock() + return cfg + case <-ctx.Done(): + return nil + } + } + c.loadDone = make(chan struct{}) + c.mu.Unlock() - start := time.Now() - cfg, err := c.store.GetQuotaConfig(ctx, c.tenantID) - if err != nil { - logger.Warn(ctx, "quota_config_cache_config_failed", - zap.String("tenant_id", c.tenantID), zap.Error(err)) - metrics.RecordTenantOperationWithOrg(c.tenantID, c.tidbCloudOrgID, "quota_config_cache", "load", "config_error", time.Since(start)) - return nil + start := now + if stale := c.snapshotCopy(); stale != nil { + if !tryAcquireQuotaConfigAsyncRefreshSlot() { + c.deferAsyncRefresh(start) + return stale + } + go func() { + defer releaseQuotaConfigAsyncRefreshSlot() + defer func() { + if recovered := recover(); recovered != nil { + logger.Error(context.Background(), "quota_config_cache_async_load_panicked", + zap.String("tenant_id", c.tenantID), zap.Any("panic", recovered)) + } + }() + c.loadConfig(context.WithoutCancel(ctx), start) + }() + return stale } - if cfg == nil { - metrics.RecordTenantOperationWithOrg(c.tenantID, c.tidbCloudOrgID, "quota_config_cache", "load", "config_empty", time.Since(start)) - return nil + return c.loadConfig(ctx, start) +} + +func tryAcquireQuotaConfigAsyncRefreshSlot() bool { + select { + case quotaConfigCacheAsyncRefreshSlots <- struct{}{}: + return true + default: + return false } +} + +func releaseQuotaConfigAsyncRefreshSlot() { + <-quotaConfigCacheAsyncRefreshSlots +} + +func (c *quotaConfigCache) deferAsyncRefresh(start time.Time) { c.mu.Lock() - if c.snapshot != nil && c.snapshot.config != nil { - existing := cloneQuotaConfigView(c.snapshot.config) - c.mu.Unlock() - metrics.RecordTenantOperationWithOrg(c.tenantID, c.tidbCloudOrgID, "quota_config_cache", "load", "raced_refresh", time.Since(start)) - return existing + if c.loadDone != nil { + c.nextRefresh = time.Now().Add(quotaConfigCacheSlotRetryInterval) + c.finishConfigLoadLocked() } - c.snapshot = "aConfigSnapshot{config: cloneQuotaConfigView(cfg), version: ""} c.mu.Unlock() - metrics.RecordTenantOperationWithOrg(c.tenantID, c.tidbCloudOrgID, "quota_config_cache", "load", "ok", time.Since(start)) - return cloneQuotaConfigView(cfg) + metrics.RecordTenantOperationWithOrg(c.tenantID, c.tidbCloudOrgID, "quota_config_cache", "load", "deferred", time.Since(start)) } -func (c *quotaConfigCache) stop() { - c.cancel() - <-c.done +func (c *quotaConfigCache) snapshotCopy() *QuotaConfigView { + c.mu.RLock() + defer c.mu.RUnlock() + return cloneQuotaConfigView(c.snapshot) } -func (c *quotaConfigCache) run(ctx context.Context) { - defer close(c.done) - ticker := time.NewTicker(quotaConfigCacheRefreshInterval) - defer ticker.Stop() - for { - select { - case <-ctx.Done(): - return - case <-ticker.C: - c.refresh(ctx) +func (c *quotaConfigCache) loadConfig(ctx context.Context, start time.Time) (result *QuotaConfigView) { + loadCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), quotaConfigCacheLoadTimeout) + defer cancel() + defer func() { + if recovered := recover(); recovered != nil { + c.finishPanickedLoad(start) + panic(recovered) } - } -} + }() -func (c *quotaConfigCache) refresh(ctx context.Context) { - start := time.Now() - version, err := c.store.GetQuotaConfigVersion(ctx, c.tenantID) + cfg, err := c.store.GetQuotaConfig(loadCtx, c.tenantID) if err != nil { - logger.Warn(ctx, "quota_config_cache_version_failed", + logger.Warn(loadCtx, "quota_config_cache_config_failed", zap.String("tenant_id", c.tenantID), zap.Error(err)) - metrics.RecordTenantOperationWithOrg(c.tenantID, c.tidbCloudOrgID, "quota_config_cache", "refresh", "version_error", time.Since(start)) - return + return c.finishFailedLoad(start, "config_error") + } + if cfg == nil { + return c.finishFailedLoad(start, "config_empty") } + c.mu.Lock() + c.snapshot = cloneQuotaConfigView(cfg) + c.nextRefresh = time.Now().Add(quotaConfigCacheRefreshDelay(quotaConfigCacheRefreshInterval)) + c.finishConfigLoadLocked() + c.mu.Unlock() + metrics.RecordTenantOperationWithOrg(c.tenantID, c.tidbCloudOrgID, "quota_config_cache", "load", "ok", time.Since(start)) + return cloneQuotaConfigView(cfg) +} - c.mu.RLock() - snapshot := c.snapshot - if snapshot != nil && snapshot.version == version { - c.mu.RUnlock() - metrics.RecordTenantOperationWithOrg(c.tenantID, c.tidbCloudOrgID, "quota_config_cache", "refresh", "unchanged", time.Since(start)) - return +func (c *quotaConfigCache) finishFailedLoad(start time.Time, result string) *QuotaConfigView { + c.mu.Lock() + c.nextRefresh = time.Now().Add(quotaConfigCacheFailureRetryDelay(quotaConfigCacheFailureRetryInterval)) + var stale *QuotaConfigView + if c.snapshot != nil { + stale = cloneQuotaConfigView(c.snapshot) } - c.mu.RUnlock() + c.finishConfigLoadLocked() + c.mu.Unlock() + metrics.RecordTenantOperationWithOrg(c.tenantID, c.tidbCloudOrgID, "quota_config_cache", "load", result, time.Since(start)) + return stale +} - cfg, err := c.store.GetQuotaConfig(ctx, c.tenantID) - if err != nil { - logger.Warn(ctx, "quota_config_cache_config_failed", - zap.String("tenant_id", c.tenantID), zap.Error(err)) - metrics.RecordTenantOperationWithOrg(c.tenantID, c.tidbCloudOrgID, "quota_config_cache", "refresh", "config_error", time.Since(start)) +func (c *quotaConfigCache) finishPanickedLoad(start time.Time) { + c.mu.Lock() + if c.loadDone == nil { + c.mu.Unlock() return } - c.mu.Lock() - c.snapshot = "aConfigSnapshot{config: cloneQuotaConfigView(cfg), version: version} + c.nextRefresh = time.Now().Add(quotaConfigCacheFailureRetryDelay(quotaConfigCacheFailureRetryInterval)) + c.finishConfigLoadLocked() c.mu.Unlock() - metrics.RecordTenantOperationWithOrg(c.tenantID, c.tidbCloudOrgID, "quota_config_cache", "refresh", "ok", time.Since(start)) + metrics.RecordTenantOperationWithOrg(c.tenantID, c.tidbCloudOrgID, "quota_config_cache", "load", "panic_error", time.Since(start)) +} + +// finishConfigLoadLocked publishes the completed load before waking cold +// waiters. c.mu must be held. +func (c *quotaConfigCache) finishConfigLoadLocked() { + done := c.loadDone + c.loadDone = nil + close(done) } type quotaUsageSnapshot struct { diff --git a/pkg/backend/quota_cache_test.go b/pkg/backend/quota_cache_test.go index 5406f2ce..a2e12598 100644 --- a/pkg/backend/quota_cache_test.go +++ b/pkg/backend/quota_cache_test.go @@ -2,6 +2,7 @@ package backend import ( "context" + "errors" "sync" "sync/atomic" "testing" @@ -11,19 +12,33 @@ import ( // cacheTestStore wraps fakeMetaQuotaStore with error injection for cache tests. type cacheTestStore struct { *fakeMetaQuotaStore - configCalls atomic.Int64 - versionCalls atomic.Int64 - usageCalls atomic.Int64 - versionErr error - configErr error - configHook func() - usageHook func() + configCalls atomic.Int64 + usageCalls atomic.Int64 + configErr error + configHook func() + configCtxHook func(context.Context) + usageHook func() } func newCacheTestStore() *cacheTestStore { return &cacheTestStore{fakeMetaQuotaStore: newFakeMetaQuotaStore()} } +func waitForQuotaConfigLoad(t *testing.T, c *quotaConfigCache) { + t.Helper() + c.mu.RLock() + done := c.loadDone + c.mu.RUnlock() + if done == nil { + return + } + select { + case <-done: + case <-time.After(300 * time.Millisecond): + t.Fatal("quota config load did not finish") + } +} + func (m *cacheTestStore) GetQuotaUsage(ctx context.Context, tenantID string) (*QuotaUsageView, error) { m.usageCalls.Add(1) if m.usageHook != nil { @@ -34,6 +49,12 @@ func (m *cacheTestStore) GetQuotaUsage(ctx context.Context, tenantID string) (*Q func (m *cacheTestStore) GetQuotaConfig(ctx context.Context, tenantID string) (*QuotaConfigView, error) { m.configCalls.Add(1) + if m.configCtxHook != nil { + m.configCtxHook(ctx) + } + if err := ctx.Err(); err != nil { + return nil, err + } if m.configErr != nil { return nil, m.configErr } @@ -43,161 +64,465 @@ func (m *cacheTestStore) GetQuotaConfig(ctx context.Context, tenantID string) (* return m.fakeMetaQuotaStore.GetQuotaConfig(ctx, tenantID) } -func (m *cacheTestStore) GetQuotaConfigVersion(ctx context.Context, tenantID string) (string, error) { - m.versionCalls.Add(1) - if m.versionErr != nil { - return "", m.versionErr +func TestQuotaConfigCacheCanceledCallerDoesNotPoisonSharedRefresh(t *testing.T) { + store := newCacheTestStore() + store.config["t1"] = &QuotaConfigView{MaxStorageBytes: 1000} + loadHasDeadline := false + store.configCtxHook = func(ctx context.Context) { + _, loadHasDeadline = ctx.Deadline() + } + c := newQuotaConfigCache("t1", "", store) + + canceledCtx, cancel := context.WithCancel(context.Background()) + cancel() + first := c.get(canceledCtx) + if first == nil || first.MaxStorageBytes != 1000 { + t.Errorf("first config = %+v, want storage 1000", first) + } + if !loadHasDeadline { + t.Error("shared load context has no deadline") + } + second := c.get(context.Background()) + if second == nil || second.MaxStorageBytes != 1000 { + t.Errorf("second config = %+v, want storage 1000", second) + } + if got := store.configCalls.Load(); got != 1 { + t.Errorf("configCalls = %d, want 1 shared load", got) } - return m.fakeMetaQuotaStore.GetQuotaConfigVersion(ctx, tenantID) } -func TestQuotaConfigCacheLazyLoad(t *testing.T) { +func TestQuotaConfigCacheRefreshDelayStaysWithinTTL(t *testing.T) { + const ttl = 30 * time.Second + for range 1000 { + delay := quotaConfigCacheRefreshDelay(ttl) + if delay < 27*time.Second || delay > ttl { + t.Errorf("refresh delay = %s, want [27s, 30s]", delay) + } + } +} + +func TestQuotaConfigCacheFailureCooldownIsFiveSeconds(t *testing.T) { store := newCacheTestStore() - store.config["t1"] = &QuotaConfigView{MaxStorageBytes: 1000} + store.configErr = errors.New("temporary metadb failure") c := newQuotaConfigCache("t1", "", store) - defer c.stop() - cfg := c.get() - if cfg != nil { - t.Fatalf("config = %+v, want nil before lazy load", cfg) + before := time.Now() + if cfg := c.get(context.Background()); cfg != nil { + t.Errorf("config = %+v, want nil", cfg) } - if got := store.versionCalls.Load(); got != 0 { - t.Fatalf("versionCalls = %d, want 0", got) + c.mu.RLock() + nextRefresh := c.nextRefresh + c.mu.RUnlock() + if delay := nextRefresh.Sub(before); delay < 4500*time.Millisecond || delay > 5500*time.Millisecond { + t.Errorf("failure cooldown = %s, want [4.5s, 5.5s]", delay) } - if got := store.configCalls.Load(); got != 0 { - t.Fatalf("configCalls = %d, want 0", got) +} + +func TestQuotaConfigCacheFailureRetryDelayUsesSymmetricJitter(t *testing.T) { + const base = 5 * time.Second + for range 2000 { + delay := quotaConfigCacheFailureRetryDelay(base) + if delay < 4500*time.Millisecond || delay > 5500*time.Millisecond { + t.Fatalf("failure retry delay = %s, want [4.5s, 5.5s]", delay) + } } +} - cfg = c.load(context.Background()) - if cfg == nil { - t.Fatal("config is nil after lazy load") +func TestQuotaConfigCachePanicDoesNotWedgeNextLoad(t *testing.T) { + store := newCacheTestStore() + store.config["t1"] = &QuotaConfigView{MaxStorageBytes: 1000} + shouldPanic := true + store.configHook = func() { + if shouldPanic { + shouldPanic = false + panic("quota config store panic") + } } - if cfg.MaxStorageBytes != 1000 { - t.Fatalf("MaxStorageBytes = %d, want 1000", cfg.MaxStorageBytes) + c := newQuotaConfigCache("t1", "", store) + + func() { + defer func() { + if recover() == nil { + t.Fatal("expected quota config load panic") + } + }() + _ = c.get(context.Background()) + }() + + c.mu.Lock() + c.nextRefresh = time.Time{} + c.mu.Unlock() + if cfg := c.get(context.Background()); cfg == nil || cfg.MaxStorageBytes != 1000 { + t.Fatalf("healthy config after panic = %+v, want storage 1000", cfg) } - if got := store.versionCalls.Load(); got != 0 { - t.Fatalf("versionCalls = %d, want 0", got) + if got := store.configCalls.Load(); got != 2 { + t.Fatalf("configCalls = %d, want panic load plus healthy reload", got) } - if got := store.configCalls.Load(); got != 1 { - t.Fatalf("configCalls = %d, want 1", got) +} + +func TestQuotaConfigCacheColdWaiterHonorsOwnDeadline(t *testing.T) { + store := newCacheTestStore() + store.config["t1"] = &QuotaConfigView{MaxStorageBytes: 1000} + started := make(chan struct{}) + release := make(chan struct{}) + var startedOnce sync.Once + var releaseOnce sync.Once + unblock := func() { releaseOnce.Do(func() { close(release) }) } + t.Cleanup(unblock) + store.configHook = func() { + startedOnce.Do(func() { close(started) }) + <-release } - if got := store.usageCalls.Load(); got != 0 { - t.Fatalf("usageCalls = %d, want 0", got) + c := newQuotaConfigCache("t1", "", store) + + leaderDone := make(chan *QuotaConfigView, 1) + go func() { leaderDone <- c.get(context.Background()) }() + <-started + + waiterCtx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + waiterDone := make(chan *QuotaConfigView, 1) + go func() { waiterDone <- c.get(waiterCtx) }() + + select { + case cfg := <-waiterDone: + if cfg != nil { + t.Errorf("waiter config = %+v, want nil before cold load completes", cfg) + } + case <-time.After(300 * time.Millisecond): + t.Error("cold waiter remained blocked after its context deadline") + } + + unblock() + if cfg := <-leaderDone; cfg == nil || cfg.MaxStorageBytes != 1000 { + t.Errorf("leader config = %+v, want storage 1000", cfg) } } -func TestQuotaConfigCacheRefreshFailOpenOnVersionError(t *testing.T) { +func TestQuotaConfigCacheWarmWaiterReturnsStaleWithoutWaiting(t *testing.T) { store := newCacheTestStore() - store.versionErr = context.DeadlineExceeded + store.config["t1"] = &QuotaConfigView{MaxStorageBytes: 1000} c := newQuotaConfigCache("t1", "", store) - defer c.stop() + if cfg := c.get(context.Background()); cfg == nil || cfg.MaxStorageBytes != 1000 { + t.Fatalf("initial config = %+v, want storage 1000", cfg) + } - c.refresh(context.Background()) - if cfg := c.get(); cfg != nil { - t.Fatalf("config = %+v, want nil", cfg) + c.mu.Lock() + c.nextRefresh = time.Time{} + c.mu.Unlock() + store.mu.Lock() + store.config["t1"] = &QuotaConfigView{MaxStorageBytes: 2000} + store.mu.Unlock() + started := make(chan struct{}) + release := make(chan struct{}) + finished := make(chan struct{}) + var startedOnce sync.Once + var releaseOnce sync.Once + unblock := func() { releaseOnce.Do(func() { close(release) }) } + t.Cleanup(unblock) + store.configHook = func() { + startedOnce.Do(func() { close(started) }) + <-release + close(finished) } - if got := store.versionCalls.Load(); got != 1 { - t.Fatalf("versionCalls = %d, want 1", got) + + leaderCtx, cancelLeader := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancelLeader() + leaderDone := make(chan *QuotaConfigView, 1) + go func() { leaderDone <- c.get(leaderCtx) }() + select { + case cfg := <-leaderDone: + if cfg == nil || cfg.MaxStorageBytes != 1000 { + t.Errorf("expired leader config = %+v, want stale storage 1000", cfg) + } + case <-time.After(100 * time.Millisecond): + t.Error("expired warm leader blocked behind the refresh") } - if got := store.configCalls.Load(); got != 0 { - t.Fatalf("configCalls = %d, want 0", got) + + <-started + waiterDone := make(chan *QuotaConfigView, 1) + go func() { waiterDone <- c.get(context.Background()) }() + select { + case cfg := <-waiterDone: + if cfg == nil || cfg.MaxStorageBytes != 1000 { + t.Errorf("warm waiter config = %+v, want stale storage 1000", cfg) + } + case <-time.After(300 * time.Millisecond): + t.Error("warm waiter blocked behind the in-flight refresh") } - if got := store.usageCalls.Load(); got != 0 { - t.Fatalf("usageCalls = %d, want 0", got) + + unblock() + select { + case <-finished: + case <-time.After(300 * time.Millisecond): + t.Fatal("async refresh did not finish") + } + if cfg := c.get(context.Background()); cfg == nil || cfg.MaxStorageBytes != 2000 { + t.Errorf("refreshed config = %+v, want storage 2000", cfg) + } +} + +func TestQuotaConfigCacheAsyncRefreshHasGlobalSlotBudget(t *testing.T) { + const tenants = defaultQuotaConfigCacheAsyncRefreshSlots + 8 + stores := make([]*cacheTestStore, tenants) + caches := make([]*quotaConfigCache, tenants) + refreshStarted := make(chan struct{}, tenants) + release := make(chan struct{}) + for i := range tenants { + store := newCacheTestStore() + store.config["t1"] = &QuotaConfigView{MaxStorageBytes: 1000} + store.configHook = func() { + refreshStarted <- struct{}{} + <-release + } + stores[i] = store + cache := newQuotaConfigCache("t1", "", store) + // The first load is cold and must not use the async refresh budget. + store.configHook = nil + if cfg := cache.get(context.Background()); cfg == nil { + t.Fatalf("initial config for tenant %d is nil", i) + } + store.configHook = func() { + refreshStarted <- struct{}{} + <-release + } + cache.mu.Lock() + cache.nextRefresh = time.Time{} + cache.mu.Unlock() + caches[i] = cache + } + + var callers sync.WaitGroup + for _, cache := range caches { + callers.Add(1) + go func(c *quotaConfigCache) { + defer callers.Done() + if cfg := c.get(context.Background()); cfg == nil || cfg.MaxStorageBytes != 1000 { + t.Errorf("expired config = %+v, want stale storage 1000", cfg) + } + }(cache) + } + callers.Wait() + + for range defaultQuotaConfigCacheAsyncRefreshSlots { + select { + case <-refreshStarted: + case <-time.After(300 * time.Millisecond): + t.Fatal("async refresh did not consume its available slots") + } + } + select { + case <-refreshStarted: + t.Fatal("async refresh exceeded the global slot budget") + default: + } + + deferred := 0 + for i, cache := range caches { + cache.mu.RLock() + loadInFlight := cache.loadDone != nil + cache.mu.RUnlock() + if stores[i].configCalls.Load() == 1 { + deferred++ + if loadInFlight { + t.Errorf("deferred tenant %d retained load ownership", i) + } + } + } + if deferred == 0 { + t.Error("expected at least one tenant refresh to be deferred") + } + + close(release) + for _, cache := range caches { + waitForQuotaConfigLoad(t, cache) + } + for i, store := range stores { + if store.configCalls.Load() > 2 { + t.Errorf("tenant %d configCalls = %d, want at most 2", i, store.configCalls.Load()) + } + } + + for i, cache := range caches { + if stores[i].configCalls.Load() != 1 { + continue + } + cache.mu.Lock() + cache.nextRefresh = time.Time{} + cache.mu.Unlock() + if cfg := cache.get(context.Background()); cfg == nil || cfg.MaxStorageBytes != 1000 { + t.Errorf("deferred tenant %d config = %+v, want stale storage 1000", i, cfg) + } + waitForQuotaConfigLoad(t, cache) + if got := stores[i].configCalls.Load(); got != 2 { + t.Errorf("deferred tenant %d configCalls = %d, want 2 after retry", i, got) + } + break } } -func TestQuotaConfigCacheRefreshOnlyLoadsConfigWhenVersionChanges(t *testing.T) { +func TestQuotaConfigCacheIsPassiveUntilFirstAccess(t *testing.T) { + previousRefreshInterval := quotaConfigCacheRefreshInterval + quotaConfigCacheRefreshInterval = 5 * time.Millisecond + t.Cleanup(func() { quotaConfigCacheRefreshInterval = previousRefreshInterval }) + store := newCacheTestStore() store.config["t1"] = &QuotaConfigView{MaxStorageBytes: 1000} c := newQuotaConfigCache("t1", "", store) - defer c.stop() - c.refresh(context.Background()) - if got := store.versionCalls.Load(); got != 1 { - t.Fatalf("versionCalls = %d, want 1", got) + // Construction must not start a polling loop or touch the store. + time.Sleep(20 * time.Millisecond) + if got := store.configCalls.Load(); got != 0 { + t.Errorf("configCalls = %d, want 0", got) + } + + cfg := c.get(context.Background()) + if cfg == nil { + t.Fatal("config is nil after first access") + } + if cfg.MaxStorageBytes != 1000 { + t.Errorf("MaxStorageBytes = %d, want 1000", cfg.MaxStorageBytes) } if got := store.configCalls.Load(); got != 1 { - t.Fatalf("configCalls = %d, want 1", got) + t.Errorf("configCalls = %d, want 1", got) + } + if got := store.usageCalls.Load(); got != 0 { + t.Errorf("usageCalls = %d, want 0", got) } +} + +func TestQuotaConfigCacheReturnsDefensiveCopy(t *testing.T) { + store := newCacheTestStore() + store.config["t1"] = &QuotaConfigView{MaxStorageBytes: 1000} + c := newQuotaConfigCache("t1", "", store) - c.refresh(context.Background()) - if got := store.versionCalls.Load(); got != 2 { - t.Fatalf("versionCalls = %d, want 2", got) + cfg := c.get(context.Background()) + if cfg == nil { + t.Fatal("config is nil") } - if got := store.configCalls.Load(); got != 1 { - t.Fatalf("configCalls = %d, want 1", got) + cfg.MaxStorageBytes = 2000 + + cached := c.get(context.Background()) + if cached == nil || cached.MaxStorageBytes != 1000 { + t.Errorf("cached config = %+v, want storage 1000", cached) } +} + +func TestQuotaConfigCacheReusesSnapshotUntilTTLExpires(t *testing.T) { + store := newCacheTestStore() + store.config["t1"] = &QuotaConfigView{MaxStorageBytes: 1000} + c := newQuotaConfigCache("t1", "", store) + if cfg := c.get(context.Background()); cfg == nil || cfg.MaxStorageBytes != 1000 { + t.Errorf("first config = %+v, want storage 1000", cfg) + } store.mu.Lock() store.config["t1"] = &QuotaConfigView{MaxStorageBytes: 2000} store.mu.Unlock() - c.refresh(context.Background()) - - cfg := c.get() - if cfg == nil { - t.Fatal("config is nil") + if cfg := c.get(context.Background()); cfg == nil || cfg.MaxStorageBytes != 1000 { + t.Errorf("cached config = %+v, want storage 1000", cfg) } - if cfg.MaxStorageBytes != 2000 { - t.Fatalf("MaxStorageBytes = %d, want 2000", cfg.MaxStorageBytes) + if got := store.configCalls.Load(); got != 1 { + t.Errorf("configCalls before expiry = %d, want 1", got) } - if got := store.versionCalls.Load(); got != 3 { - t.Fatalf("versionCalls = %d, want 3", got) + + c.mu.Lock() + c.nextRefresh = time.Time{} + c.mu.Unlock() + if cfg := c.get(context.Background()); cfg == nil || cfg.MaxStorageBytes != 1000 { + t.Errorf("expired config = %+v, want stale storage 1000", cfg) } - if got := store.configCalls.Load(); got != 2 { - t.Fatalf("configCalls = %d, want 2", got) + waitForQuotaConfigLoad(t, c) + if cfg := c.get(context.Background()); cfg == nil || cfg.MaxStorageBytes != 2000 { + t.Errorf("refreshed config = %+v, want storage 2000", cfg) } - if got := store.usageCalls.Load(); got != 0 { - t.Fatalf("usageCalls = %d, want 0", got) + if got := store.configCalls.Load(); got != 2 { + t.Errorf("configCalls after expiry = %d, want 2", got) } } -func TestQuotaConfigCacheLazyLoadDoesNotOverwriteRefreshedSnapshot(t *testing.T) { +func TestQuotaConfigCacheCoalescesConcurrentExpiredAccess(t *testing.T) { store := newCacheTestStore() store.config["t1"] = &QuotaConfigView{MaxStorageBytes: 1000} c := newQuotaConfigCache("t1", "", store) - defer c.stop() + if cfg := c.get(context.Background()); cfg == nil { + t.Fatal("initial config is nil") + } + c.mu.Lock() + c.nextRefresh = time.Time{} + c.mu.Unlock() + started := make(chan struct{}) + release := make(chan struct{}) + var once sync.Once store.configHook = func() { - c.mu.Lock() - c.snapshot = "aConfigSnapshot{ - config: &QuotaConfigView{MaxStorageBytes: 2000}, - version: "new-version", - } - c.mu.Unlock() + once.Do(func() { close(started) }) + <-release } - - cfg := c.load(context.Background()) - if cfg == nil { - t.Fatal("config is nil") + const callers = 32 + results := make(chan *QuotaConfigView, callers) + for range callers { + go func() { results <- c.get(context.Background()) }() } - if cfg.MaxStorageBytes != 2000 { - t.Fatalf("lazy load config = %d, want refreshed 2000", cfg.MaxStorageBytes) + <-started + close(release) + for range callers { + if cfg := <-results; cfg == nil || cfg.MaxStorageBytes != 1000 { + t.Errorf("concurrent config = %+v, want storage 1000", cfg) + } } - cached := c.get() - if cached == nil || cached.MaxStorageBytes != 2000 { - t.Fatalf("cached config = %+v, want refreshed 2000", cached) + if got := store.configCalls.Load(); got != 2 { + t.Errorf("configCalls = %d, want initial load plus one coalesced refresh", got) } } -func TestQuotaConfigCacheLazyLoadReturnsDefensiveCopy(t *testing.T) { +func TestQuotaConfigCacheRefreshFailureReturnsStaleAndUsesRetryCooldown(t *testing.T) { store := newCacheTestStore() store.config["t1"] = &QuotaConfigView{MaxStorageBytes: 1000} c := newQuotaConfigCache("t1", "", store) - defer c.stop() - cfg := c.load(context.Background()) - if cfg == nil { - t.Fatal("config is nil") + first := c.get(context.Background()) + if first == nil { + t.Fatal("initial config is nil") } - cfg.MaxStorageBytes = 2000 + c.mu.Lock() + c.nextRefresh = time.Time{} + c.mu.Unlock() + store.configErr = errors.New("temporary metadb failure") + + stale := c.get(context.Background()) + if stale == nil || stale.MaxStorageBytes != 1000 { + t.Errorf("stale config = %+v, want storage 1000", stale) + } + waitForQuotaConfigLoad(t, c) + if got := store.configCalls.Load(); got != 2 { + t.Errorf("configCalls after failed refresh = %d, want 2", got) + } + stale = c.get(context.Background()) + if stale == nil || stale.MaxStorageBytes != 1000 { + t.Errorf("cooldown config = %+v, want storage 1000", stale) + } + if got := store.configCalls.Load(); got != 2 { + t.Errorf("configCalls during failure cooldown = %d, want 2", got) + } +} + +func TestQuotaConfigCacheInitialFailureUsesRetryCooldown(t *testing.T) { + store := newCacheTestStore() + store.configErr = errors.New("temporary metadb failure") + c := newQuotaConfigCache("t1", "", store) - cached := c.get() - if cached == nil { - t.Fatal("cached config is nil") + if cfg := c.get(context.Background()); cfg != nil { + t.Errorf("config = %+v, want nil on initial failure", cfg) + } + if cfg := c.get(context.Background()); cfg != nil { + t.Errorf("config during cooldown = %+v, want nil", cfg) + } + if got := store.configCalls.Load(); got != 1 { + t.Errorf("configCalls during failure cooldown = %d, want 1", got) } - if cached.MaxStorageBytes != 1000 { - t.Fatalf("cached MaxStorageBytes = %d, want 1000", cached.MaxStorageBytes) + if got := store.usageCalls.Load(); got != 0 { + t.Errorf("usageCalls = %d, want 0", got) } } @@ -494,10 +819,3 @@ func TestQuotaPendingDeltasCacheRemovesPositiveRaceDeltasOnClearAndExpire(t *tes t.Fatalf("positive deltas after expire = %+v, want zero", got) } } - -func TestQuotaConfigCacheStop(t *testing.T) { - store := newCacheTestStore() - c := newQuotaConfigCache("t1", "", store) - c.stop() - // Should not panic or block. -} diff --git a/pkg/backend/quota_integration_test.go b/pkg/backend/quota_integration_test.go index 562091d0..7fd263ec 100644 --- a/pkg/backend/quota_integration_test.go +++ b/pkg/backend/quota_integration_test.go @@ -29,6 +29,19 @@ func newServerQuotaBackend(t *testing.T, opts Options) (*Dat9Backend, *fakeMetaQ return b, fake } +func reloadQuotaConfigForTest(t *testing.T, b *Dat9Backend, ctx context.Context) { + t.Helper() + if b.quotaConfigCache == nil { + t.Fatal("quota config cache is nil") + } + b.quotaConfigCache.mu.Lock() + b.quotaConfigCache.nextRefresh = time.Time{} + b.quotaConfigCache.mu.Unlock() + if cfg := b.quotaConfigCache.get(ctx); cfg == nil { + t.Fatal("reload quota config returned nil") + } +} + func waitForFakeCentralLLMUsage(t *testing.T, fake *fakeMetaQuotaStore, tenantID string, wantMonthly int64, wantUsageLen int) { t.Helper() deadline := time.Now().Add(2 * time.Second) @@ -117,7 +130,7 @@ func TestServerQuotaRejectsOverFileSizeLimit(t *testing.T) { fake.mu.Lock() fake.config["tenant-a"].MaxFileSizeBytes = 4 fake.mu.Unlock() - b.quotaConfigCache.refresh(ctx) + reloadQuotaConfigForTest(t, b, ctx) if _, err := b.WriteCtx(ctx, "/too-large.txt", []byte("12345"), 0, filesystem.WriteFlagCreate); !errors.Is(err, ErrFileSizeQuotaExceeded) { t.Fatalf("write error = %v, want ErrFileSizeQuotaExceeded", err) @@ -141,7 +154,7 @@ func TestCreateIfAbsentExistingPathReturnsConflictBeforeFileSizeQuota(t *testing fake.mu.Lock() fake.config["tenant-a"].MaxFileSizeBytes = 4 fake.mu.Unlock() - b.quotaConfigCache.refresh(ctx) + reloadQuotaConfigForTest(t, b, ctx) if _, _, err := b.WriteCtxIfRevisionWithTagsResult(ctx, "/size-exists.txt", []byte("12345"), 0, filesystem.WriteFlagCreate|filesystem.WriteFlagTruncate, 0, nil, ""); !errors.Is(err, datastore.ErrRevisionConflict) { t.Fatalf("duplicate create-if-absent error = %v, want ErrRevisionConflict", err) @@ -161,7 +174,7 @@ func TestServerModeBudgetGateWritesCentralOnly(t *testing.T) { fake.mu.Lock() fake.config["tenant-a"].MaxMonthlyCostMC = 100 fake.mu.Unlock() - b.quotaConfigCache.refresh(context.Background()) + reloadQuotaConfigForTest(t, b, context.Background()) b.recordImageExtractUsage("task-server-budget", ImageExtractUsage{ PromptTokens: 120, diff --git a/pkg/backend/quota_migration_test.go b/pkg/backend/quota_migration_test.go index c27a662b..92cda863 100644 --- a/pkg/backend/quota_migration_test.go +++ b/pkg/backend/quota_migration_test.go @@ -121,15 +121,6 @@ func (f *fakeMetaQuotaStore) GetQuotaConfig(ctx context.Context, tenantID string }, nil } -func (f *fakeMetaQuotaStore) GetQuotaConfigVersion(ctx context.Context, tenantID string) (string, error) { - f.mu.Lock() - defer f.mu.Unlock() - if cfg, ok := f.config[tenantID]; ok { - return fmt.Sprintf("v3:%d:%d:%d:%d:%d:%d", cfg.MaxStorageBytes, cfg.MaxFileSizeBytes, cfg.MaxFileCount, cfg.MaxMediaLLMFiles, cfg.MaxVideoLLMFiles, cfg.MaxMonthlyCostMC), nil - } - return "", nil -} - func (f *fakeMetaQuotaStore) GetQuotaUsage(ctx context.Context, tenantID string) (*QuotaUsageView, error) { f.mu.Lock() defer f.mu.Unlock() diff --git a/pkg/backend/quota_store.go b/pkg/backend/quota_store.go index 1d24df5c..90199ad3 100644 --- a/pkg/backend/quota_store.go +++ b/pkg/backend/quota_store.go @@ -21,7 +21,6 @@ const defaultTenantMetricTiDBCloudOrgID = "guest" type MetaQuotaStore interface { // Config GetQuotaConfig(ctx context.Context, tenantID string) (*QuotaConfigView, error) - GetQuotaConfigVersion(ctx context.Context, tenantID string) (string, error) // Counters GetQuotaUsage(ctx context.Context, tenantID string) (*QuotaUsageView, error) diff --git a/pkg/meta/quota.go b/pkg/meta/quota.go index cba4310b..29e3e48a 100644 --- a/pkg/meta/quota.go +++ b/pkg/meta/quota.go @@ -169,35 +169,6 @@ func (s *Store) GetQuotaConfig(ctx context.Context, tenantID string) (*QuotaConf return cfg, nil } -// GetQuotaConfigVersion returns a lightweight content token for a tenant's -// explicit storage quota config. An empty token means no storage-relevant row -// exists and callers should use GetQuotaConfig's default storage config. The -// token is derived from the effective config values instead of updated_at so -// updates inside the same timestamp tick cannot hide a real config change from -// cache invalidation. -func (s *Store) GetQuotaConfigVersion(ctx context.Context, tenantID string) (string, error) { - start := time.Now() - var err error - defer observeMeta(ctx, "get_quota_config_version", start, &err) - - var maxStorageBytes, maxFileSizeBytes, maxFileCount, maxMediaLLMFiles, maxVideoLLMFiles, maxMonthlyCostMC int64 - err = s.db.QueryRowContext(ctx, - `SELECT max_storage_bytes, max_file_size_bytes, max_file_count, - max_media_llm_files, max_video_llm_files, max_monthly_cost_mc - FROM tenant_quota_config WHERE tenant_id = ?`, tenantID, - ).Scan(&maxStorageBytes, &maxFileSizeBytes, &maxFileCount, &maxMediaLLMFiles, &maxVideoLLMFiles, &maxMonthlyCostMC) - if err == sql.ErrNoRows { - logger.Info(ctx, "quota_config_not_found_using_defaults", - zap.String("tenant_id", tenantID)) - err = nil - return "", nil - } - if err != nil { - return "", fmt.Errorf("get quota config version for tenant %q: %w", tenantID, err) - } - return fmt.Sprintf("v3:%d:%d:%d:%d:%d:%d", maxStorageBytes, maxFileSizeBytes, maxFileCount, maxMediaLLMFiles, maxVideoLLMFiles, maxMonthlyCostMC), nil -} - // SetQuotaConfig upserts per-tenant quota configuration. func (s *Store) SetQuotaConfig(ctx context.Context, cfg *QuotaConfig) error { start := time.Now() diff --git a/pkg/meta/quota_setting_test.go b/pkg/meta/quota_setting_test.go index 07416432..a7af9eea 100644 --- a/pkg/meta/quota_setting_test.go +++ b/pkg/meta/quota_setting_test.go @@ -87,7 +87,7 @@ func TestGetQuotaConfigUsesConfiguredDefaultStorageBytes(t *testing.T) { } } -func TestQuotaConfigStoresTiDBCloudSpendingLimitWithoutStorageVersion(t *testing.T) { +func TestQuotaConfigSpendingLimitPatchMaterializesStorageDefaults(t *testing.T) { originalFileSizeDefault := DefaultMaxFileSizeBytes() defer SetDefaultMaxFileSizeBytes(originalFileSizeDefault) @@ -99,7 +99,7 @@ func TestQuotaConfigStoresTiDBCloudSpendingLimitWithoutStorageVersion(t *testing t.Fatal(err) } if cfg.TiDBCloudSpendingLimit != nil { - t.Fatalf("default spending limit = %#v, want nil", cfg.TiDBCloudSpendingLimit) + t.Errorf("default spending limit = %#v, want nil", cfg.TiDBCloudSpendingLimit) } zero := int64(0) @@ -111,10 +111,10 @@ func TestQuotaConfigStoresTiDBCloudSpendingLimitWithoutStorageVersion(t *testing t.Fatal(err) } if cfg.TiDBCloudSpendingLimit == nil || *cfg.TiDBCloudSpendingLimit != 0 { - t.Fatalf("spending limit = %#v, want 0", cfg.TiDBCloudSpendingLimit) + t.Errorf("spending limit = %#v, want 0", cfg.TiDBCloudSpendingLimit) } if cfg.MaxStorageBytes != DefaultMaxStorageBytes() || cfg.MaxFileSizeBytes != DefaultMaxFileSizeBytes() || cfg.MaxFileCount != 0 { - t.Fatalf("storage quota fields = %#v, want defaults", cfg) + t.Errorf("storage quota fields = %#v, want defaults", cfg) } SetDefaultMaxFileSizeBytes(originalFileSizeDefault + 1) cfg, err = s.GetQuotaConfig(ctx, "tenant-spending-only") @@ -122,16 +122,8 @@ func TestQuotaConfigStoresTiDBCloudSpendingLimitWithoutStorageVersion(t *testing t.Fatal(err) } if cfg.MaxFileSizeBytes != originalFileSizeDefault { - t.Fatalf("materialized MaxFileSizeBytes = %d after default changed, want %d", cfg.MaxFileSizeBytes, originalFileSizeDefault) + t.Errorf("materialized MaxFileSizeBytes = %d after default changed, want %d", cfg.MaxFileSizeBytes, originalFileSizeDefault) } - version, err := s.GetQuotaConfigVersion(ctx, "tenant-spending-only") - if err != nil { - t.Fatal(err) - } - if version == "" { - t.Fatalf("storage quota version should be non-empty when config row exists") - } - updated := int64(123) if err := s.SetQuotaConfigPatch(ctx, "tenant-spending-only", QuotaConfigPatch{TiDBCloudSpendingLimit: &updated}); err != nil { t.Fatal(err) @@ -141,7 +133,7 @@ func TestQuotaConfigStoresTiDBCloudSpendingLimitWithoutStorageVersion(t *testing t.Fatal(err) } if cfg.TiDBCloudSpendingLimit == nil || *cfg.TiDBCloudSpendingLimit != updated { - t.Fatalf("updated spending limit = %#v, want %d", cfg.TiDBCloudSpendingLimit, updated) + t.Errorf("updated spending limit = %#v, want %d", cfg.TiDBCloudSpendingLimit, updated) } checkedAt := time.Now().UTC() if err := s.SetQuotaConfigPatch(ctx, "tenant-spending-only", QuotaConfigPatch{TiDBCloudSpendingLimitCheckedAt: &checkedAt}); err != nil { @@ -152,7 +144,7 @@ func TestQuotaConfigStoresTiDBCloudSpendingLimitWithoutStorageVersion(t *testing t.Fatal(err) } if cfg.TiDBCloudSpendingLimitCheckedAt == nil { - t.Fatal("spending limit checked_at = nil, want timestamp") + t.Error("spending limit checked_at = nil, want timestamp") } } @@ -176,48 +168,6 @@ func TestGetQuotaConfigUsesDefaultFileSizeForExistingZeroRow(t *testing.T) { } } -func TestGetQuotaConfigVersion(t *testing.T) { - s := newControlStore(t) - ctx := context.Background() - - version, err := s.GetQuotaConfigVersion(ctx, "tenant-without-config") - if err != nil { - t.Fatal(err) - } - if version != "" { - t.Fatalf("version for missing config = %q, want empty", version) - } - - if err := s.SetQuotaConfig(ctx, &QuotaConfig{ - TenantID: "tenant-with-config", - MaxStorageBytes: 123, - MaxFileSizeBytes: 234, - MaxFileCount: 345, - MaxMediaLLMFiles: 456, - MaxVideoLLMFiles: 567, - MaxMonthlyCostMC: 789, - }); err != nil { - t.Fatal(err) - } - version, err = s.GetQuotaConfigVersion(ctx, "tenant-with-config") - if err != nil { - t.Fatal(err) - } - if version == "" { - t.Fatal("version for explicit config is empty") - } - if err := s.SetQuotaStorageBytes(ctx, "tenant-with-config", 321); err != nil { - t.Fatal(err) - } - nextVersion, err := s.GetQuotaConfigVersion(ctx, "tenant-with-config") - if err != nil { - t.Fatal(err) - } - if nextVersion == version { - t.Fatalf("version after config value change = %q, want different from %q", nextVersion, version) - } -} - func TestSetQuotaStorageBytesUpdatesStorageOnly(t *testing.T) { s := newControlStore(t) ctx := context.Background() diff --git a/pkg/server/quota_test.go b/pkg/server/quota_test.go index edacd140..2f4255c2 100644 --- a/pkg/server/quota_test.go +++ b/pkg/server/quota_test.go @@ -408,6 +408,17 @@ func newQuotaRuntime(t *testing.T, provider string) *quotaRuntime { return newQuotaRuntimeWithOptions(t, provider, quotaRuntimeOptions{}) } +func quotaConfigRowExists(t *testing.T, rt *quotaRuntime) bool { + t.Helper() + var exists bool + if err := rt.meta.DB().QueryRowContext(context.Background(), + `SELECT EXISTS(SELECT 1 FROM tenant_quota_config WHERE tenant_id = ?)`, rt.tenantID, + ).Scan(&exists); err != nil { + t.Fatal(err) + } + return exists +} + func newQuotaRuntimeWithOptions(t *testing.T, provider string, opts quotaRuntimeOptions) *quotaRuntime { t.Helper() db := newTenantDeleteDBInfo(t) @@ -1089,35 +1100,31 @@ func TestQuotaSetSpendingLimitOnlyPersistsSpendingLimitConfig(t *testing.T) { } calls := rt.prov.callsSnapshot() if len(calls) != 2 || calls[0] != "mark" || calls[1] != "update" { - t.Fatalf("calls = %#v, want mark before update", calls) + t.Errorf("calls = %#v, want mark before update", calls) } lastOptions := rt.prov.lastOptionsSnapshot() if lastOptions.TiDBCloudSpendingLimitMonthly == nil || *lastOptions.TiDBCloudSpendingLimitMonthly != spendingLimit { - t.Fatalf("spending limit option = %#v, want %d", lastOptions.TiDBCloudSpendingLimitMonthly, spendingLimit) + t.Errorf("spending limit option = %#v, want %d", lastOptions.TiDBCloudSpendingLimitMonthly, spendingLimit) } cfg, err := rt.meta.GetQuotaConfig(context.Background(), rt.tenantID) if err != nil { t.Fatal(err) } if cfg.TiDBCloudSpendingLimit == nil || *cfg.TiDBCloudSpendingLimit != spendingLimit { - t.Fatalf("persisted spending limit = %#v, want %d", cfg.TiDBCloudSpendingLimit, spendingLimit) + t.Errorf("persisted spending limit = %#v, want %d", cfg.TiDBCloudSpendingLimit, spendingLimit) } if cfg.MaxStorageBytes != meta.DefaultMaxStorageBytes() || cfg.MaxFileSizeBytes != meta.DefaultMaxFileSizeBytes() || cfg.MaxFileCount != 0 { - t.Fatalf("storage quota fields = %#v, want defaults", cfg) - } - version, err := rt.meta.GetQuotaConfigVersion(context.Background(), rt.tenantID) - if err != nil { - t.Fatal(err) + t.Errorf("storage quota fields = %#v, want defaults", cfg) } - if version == "" { - t.Fatalf("storage quota config version should be non-empty when config row exists") + if !quotaConfigRowExists(t, rt) { + t.Error("quota config row was not persisted") } var out quotaResponse if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { t.Fatal(err) } if out.Config.TiDBCloudSpendingLimit == nil || *out.Config.TiDBCloudSpendingLimit != spendingLimit { - t.Fatalf("config = %#v", out.Config) + t.Errorf("config = %#v", out.Config) } } @@ -1140,12 +1147,8 @@ func TestQuotaSetRejectsDrive9KeyWithoutTiDBCloudCredentials(t *testing.T) { if got := rt.prov.markCalls.Load(); got != 0 { t.Fatalf("mark calls = %d, want 0", got) } - version, err := rt.meta.GetQuotaConfigVersion(context.Background(), rt.tenantID) - if err != nil { - t.Fatal(err) - } - if version != "" { - t.Fatalf("quota config version = %q, want empty", version) + if quotaConfigRowExists(t, rt) { + t.Error("quota config row was written for a rejected request") } } @@ -1366,12 +1369,8 @@ func TestQuotaSetMapsTiDBCloudCredentialErrorsWithoutWritingConfig(t *testing.T) if resp.StatusCode != tc.wantStatus { t.Fatalf("status = %d, want %d", resp.StatusCode, tc.wantStatus) } - version, err := rt.meta.GetQuotaConfigVersion(context.Background(), rt.tenantID) - if err != nil { - t.Fatal(err) - } - if version != "" { - t.Fatalf("quota config version = %q, want empty", version) + if quotaConfigRowExists(t, rt) { + t.Error("quota config row was written after TiDB Cloud credential failure") } if got := rt.prov.markCalls.Load(); got != 1 { t.Fatalf("mark calls = %d, want 1", got) diff --git a/pkg/tenant/pool.go b/pkg/tenant/pool.go index e274bdc7..2352abef 100644 --- a/pkg/tenant/pool.go +++ b/pkg/tenant/pool.go @@ -66,9 +66,11 @@ type PoolConfig struct { // IsLeader() snapshot. LeaderChecker LeaderChecker - // IdleTimeout controls how long a cached backend can stay in the warm - // cache without any activity before the idle reaper evicts it. 0 - // disables idle eviction (LRU capacity eviction still applies). + // IdleTimeout controls how long a standalone cached backend can stay in the + // warm cache without activity before the idle reaper evicts it. Shared + // tenant entries intentionally bypass this TTL and are reclaimed by the + // capacity LRU instead. 0 disables standalone idle eviction; the shared + // physical-DB handle reaper remains active. // // "Activity" means any access through Get, Acquire, or S3Backend — // including foreground user requests (HTTP/FUSE), tenant-specific @@ -83,8 +85,9 @@ type PoolConfig struct { // per due attempt. IdleTimeout time.Duration - // IdleReapInterval is how often the idle reaper scans for idle backends. - // Defaults to defaultTenantPoolIdleReapInterval when IdleTimeout > 0. + // IdleReapInterval is how often the reaper scans standalone entries and + // unreferenced shared physical-DB handles. It defaults to + // defaultTenantPoolIdleReapInterval when not set. IdleReapInterval time.Duration // SharedDBForcePlaintext, when true, opens shared-schema DB handles without @@ -248,7 +251,7 @@ func NewPool(cfg PoolConfig, enc encrypt.Encryptor) *Pool { } idleTimeout := cfg.IdleTimeout reapInterval := cfg.IdleReapInterval - if reapInterval <= 0 && idleTimeout > 0 { + if reapInterval <= 0 { reapInterval = defaultTenantPoolIdleReapInterval } metrics.RecordGauge("tenant_pool", "cached_backends", 0) @@ -568,7 +571,6 @@ func (p *Pool) AcquireCached(t *meta.Tenant) (b *backend.Dat9Backend, release fu return nil, nil, false } e.refs++ - p.order.MoveToFront(e.elem) p.mu.Unlock() metrics.RecordOperation("tenant_pool", "cache_lookup", "hit", 0) metrics.RecordTenantOperationWithOrg(t.ID, e.backend.TiDBCloudOrgID(), "user_db_access", "acquire_cached", "hit", 0) @@ -692,11 +694,11 @@ func withTenantPoolDrainTimeout(ctx context.Context) (context.Context, context.C return context.WithTimeout(ctx, defaultTenantPoolDrainTimeout) } -// Start launches the idle reaper goroutine if IdleTimeout is configured. -// Safe to call on a pool with IdleTimeout=0 (no-op). The reaper is stopped -// by Close. +// Start launches the reaper goroutine. IdleTimeout=0 disables standalone +// tenant eviction, but shared physical-DB handles still need this reaper. +// The reaper is stopped by Close. func (p *Pool) Start(ctx context.Context) { - if p == nil || p.idleTimeout <= 0 { + if p == nil || p.reapInterval <= 0 { return } workerCtx, cancel := context.WithCancel(ctx) @@ -725,24 +727,30 @@ func (p *Pool) reapLoop(ctx context.Context) { } func (p *Pool) reapOnce(ctx context.Context) { - if p.idleTimeout <= 0 { - return - } now := time.Now() var toClose []*entry - p.mu.Lock() - for _, e := range p.items { - if e.retired || e.refs > 0 { - continue - } - if now.Sub(e.lastUsed) > p.idleTimeout { - if removed := p.removeLocked(e.elem, "idle"); removed != nil { - toClose = append(toClose, removed) + if p.idleTimeout > 0 { + p.mu.Lock() + for _, e := range p.items { + if e.retired || e.refs > 0 { + continue + } + // Shared tenant backends are lightweight fs_id-scoped views over a + // separately managed physical DB handle. Keep them warm until the + // capacity LRU evicts them; applying the standalone idle TTL here turns + // stable high-cardinality shared traffic into continuous cold opens. + if e.sharedDBID > 0 { + continue + } + if now.Sub(e.lastUsed) > p.idleTimeout { + if removed := p.removeLocked(e.elem, "idle"); removed != nil { + toClose = append(toClose, removed) + } } } + p.recordCachedBackendCountLocked() + p.mu.Unlock() } - p.recordCachedBackendCountLocked() - p.mu.Unlock() for _, retired := range toClose { p.closeEntry(retired) } diff --git a/pkg/tenant/pool_test.go b/pkg/tenant/pool_test.go index 4379a822..03638797 100644 --- a/pkg/tenant/pool_test.go +++ b/pkg/tenant/pool_test.go @@ -182,6 +182,29 @@ func TestSharedDBHandleIdleReapUsesLongerTTLAndSkipsReferencedHandles(t *testing } } +func TestSharedDBHandleIdleReapRunsWhenTenantIdleEvictionDisabled(t *testing.T) { + p := NewPool(PoolConfig{IdleTimeout: 0}, nil) + t.Cleanup(p.Close) + db, err := sql.Open("mysql", testDSN) + if err != nil { + t.Fatal(err) + } + if err := db.Ping(); err != nil { + t.Fatal(err) + } + p.sharedDBs[1] = db + p.sharedDBLastUsed[1] = time.Now().Add(-defaultSharedDBHandleIdleTTL - time.Second) + + p.reapOnce(context.Background()) + + if _, ok := p.sharedDBs[1]; ok { + t.Fatal("expired unreferenced shared handle remains cached when tenant idle eviction is disabled") + } + if err := db.Ping(); err == nil { + t.Fatal("expired unreferenced shared handle remains open when tenant idle eviction is disabled") + } +} + func registerSharedDBForCacheMetrics(t *testing.T, orgID string) (*meta.Store, *Pool, int64, string) { t.Helper() metaStore, err := meta.OpenContext(context.Background(), testDSN) @@ -1187,6 +1210,118 @@ func TestIdleEviction(t *testing.T) { } } +func TestIdleEvictionSkipsSharedTenantEntry(t *testing.T) { + pool, tenant := newTestPoolAndTenantWithConfig(t, PoolConfig{ + MaxTenants: 2, + IdleTimeout: time.Minute, + }, "tenant-shared-idle") + ctx := context.Background() + + b, release, err := pool.Acquire(ctx, tenant) + if err != nil { + t.Fatal(err) + } + store := b.Store() + release() + + pool.mu.Lock() + e := pool.items[tenant.ID] + e.sharedDBID = 123 + e.lastUsed = time.Now().Add(-2 * time.Minute) + pool.mu.Unlock() + + pool.reapOnce(ctx) + + assertStoreOpen(t, store) + if _, ok := pool.items[tenant.ID]; !ok { + t.Fatal("shared tenant entry was removed by idle reaper") + } +} + +func TestCapacityEvictionStillRemovesSharedTenantEntry(t *testing.T) { + pool, first := newTestPoolAndTenant(t, 1, "tenant-shared-capacity-first") + ctx := context.Background() + + b, release, err := pool.Acquire(ctx, first) + if err != nil { + t.Fatal(err) + } + firstStore := b.Store() + release() + + pool.mu.Lock() + pool.items[first.ID].sharedDBID = 123 + pool.mu.Unlock() + + second := cloneTenantForID(t, pool, first, "tenant-shared-capacity-second") + _, releaseSecond, err := pool.Acquire(ctx, second) + if err != nil { + t.Fatal(err) + } + releaseSecond() + + assertStoreClosed(t, firstStore) + if _, ok := pool.items[first.ID]; ok { + t.Fatal("shared tenant entry remained after capacity eviction") + } +} + +func TestAcquireCachedDoesNotRefreshSharedCapacityLRU(t *testing.T) { + pool, busy := newTestPoolAndTenant(t, 2, "tenant-shared-lru-busy") + ctx := context.Background() + + busyBackend, releaseBusy, err := pool.Acquire(ctx, busy) + if err != nil { + t.Fatal(err) + } + busyStore := busyBackend.Store() + releaseBusy() + + idle := cloneTenantForID(t, pool, busy, "tenant-shared-lru-idle") + idleBackend, releaseIdle, err := pool.Acquire(ctx, idle) + if err != nil { + t.Fatal(err) + } + idleStore := idleBackend.Store() + releaseIdle() + + pool.mu.Lock() + pool.items[busy.ID].sharedDBID = 123 + pool.items[idle.ID].sharedDBID = 123 + pool.mu.Unlock() + + // Foreground traffic makes busy the most recently used entry. + _, releaseBusy, err = pool.Acquire(ctx, busy) + if err != nil { + t.Fatal(err) + } + releaseBusy() + + // The safety-net scan may pin idle, but must not rewrite capacity LRU order. + _, releaseCached, ok := pool.AcquireCached(idle) + if !ok { + t.Fatal("expected AcquireCached to hit idle shared tenant") + } + releaseCached() + + third := cloneTenantForID(t, pool, busy, "tenant-shared-lru-third") + _, releaseThird, err := pool.Acquire(ctx, third) + if err != nil { + t.Fatal(err) + } + releaseThird() + + pool.mu.Lock() + _, busyCached := pool.items[busy.ID] + _, idleCached := pool.items[idle.ID] + pool.mu.Unlock() + if !busyCached || idleCached { + t.Errorf("capacity LRU after safety-net scan: busy cached=%t idle cached=%t, want true false", busyCached, idleCached) + } + assertStoreOpen(t, busyStore) + assertStoreClosed(t, idleStore) +} + func TestCloseEntryPreservesLiveTenantCounters(t *testing.T) { const tenantID = "tenant-live-counter-close-entry" metrics.DeleteTenantCounters(tenantID) @@ -1341,14 +1476,15 @@ func TestIdleEvictionDisabled(t *testing.T) { } } -func TestIdleEvictionStartNoOpWhenDisabled(t *testing.T) { +func TestIdleEvictionStartKeepsSharedHandleReaperWhenDisabled(t *testing.T) { pool := NewPool(PoolConfig{ MaxTenants: 2, IdleTimeout: 0, }, nil) pool.Start(context.Background()) - if pool.reapStop != nil { - t.Fatal("expected reapStop to be nil when IdleTimeout=0") + t.Cleanup(pool.Close) + if pool.reapStop == nil { + t.Fatal("expected shared handle reaper to run when IdleTimeout=0") } } diff --git a/pkg/tenant/quota_adapter.go b/pkg/tenant/quota_adapter.go index 139d53cf..8ce18f05 100644 --- a/pkg/tenant/quota_adapter.go +++ b/pkg/tenant/quota_adapter.go @@ -41,10 +41,6 @@ func (a *metaQuotaAdapter) GetQuotaConfig(ctx context.Context, tenantID string) }, nil } -func (a *metaQuotaAdapter) GetQuotaConfigVersion(ctx context.Context, tenantID string) (string, error) { - return a.s.GetQuotaConfigVersion(ctx, tenantID) -} - func (a *metaQuotaAdapter) GetQuotaUsage(ctx context.Context, tenantID string) (*backend.QuotaUsageView, error) { u, err := a.s.GetQuotaUsage(ctx, tenantID) if err != nil {