diff --git a/autocleanup.go b/autocleanup.go index 0b1e4c8..72add80 100644 --- a/autocleanup.go +++ b/autocleanup.go @@ -46,6 +46,12 @@ func NewBackgroundTask(name string, interval time.Duration, taskFunc func(), log if len(wg) > 0 { externalWG = wg[0] } + // run() uses time.NewTicker(interval), which panics for d <= 0 inside + // the started goroutine (autocleanup.go:186), crashing the whole + // process on a misconfigured interval. Normalize here (R149). + if interval <= 0 { + interval = time.Minute + } return &BackgroundTask{ name: name, interval: interval, @@ -88,8 +94,20 @@ func (bt *BackgroundTask) Start() { return } - // Reserve the task slot immediately when starting - registry.cb.OnTaskStart(bt.name) + // Reserve the task slot immediately when starting. OnTaskStart + // is the authoritative concurrency gate (R165): it returns false + // if maxConcurrent is already reached, in which case this task + // must not launch run() (its slot was not taken). + if !registry.cb.OnTaskStart(bt.name) { + if bt.logger != nil { + bt.logger.Debugf("Cannot start task %s: concurrency limit reached, slot not acquired", bt.name) + } + // Close doneChan since the task won't run + if atomic.CompareAndSwapInt32(&bt.doneClosed, 0, 1) { + close(bt.doneChan) + } + return + } atomic.StoreInt32(&bt.started, 1) bt.internalWG.Add(1) @@ -114,6 +132,14 @@ func (bt *BackgroundTask) Stop() { if atomic.CompareAndSwapInt32(&bt.doneClosed, 0, 1) { close(bt.doneChan) } + // Also close stopChan (R117): a concurrent Start may be in-flight + // and about to launch run(). Signaling stopChan lets it exit at + // the next stop check instead of leaking the goroutine and its + // circuit-breaker slot (taken by OnTaskStart but never released). + func() { + defer func() { _ = recover() }() // safe if already closed + close(bt.stopChan) + }() return } @@ -134,15 +160,18 @@ func (bt *BackgroundTask) Stop() { // This avoids the race condition with WaitGroup select { case <-bt.doneChan: - // Normal completion + // Normal completion: the goroutine has returned, so it is safe + // to wait on the WaitGroup. + bt.internalWG.Wait() case <-time.After(5 * time.Second): + // The task goroutine is still running (its taskFunc is blocked or + // long-lived). Do NOT wait on the WaitGroup here — that would + // block Stop() (and shutdown) for as long as the task runs. + // Return now; the goroutine is reaped when it completes. if bt.logger != nil { bt.logger.Errorf("Timeout waiting for background task %s to stop", bt.name) } } - - // Wait for the internal WaitGroup synchronously after doneChan signals - bt.internalWG.Wait() }) } @@ -345,9 +374,30 @@ func (cb *TaskCircuitBreaker) CanCreateTask(taskName string) error { } } -// OnTaskStart records a task starting execution -func (cb *TaskCircuitBreaker) OnTaskStart(taskName string) { - atomic.AddInt32(&cb.concurrentTasks, 1) +// OnTaskStart records a task starting execution. It atomically reserves a +// concurrency slot and returns whether one was available. The reservation +// is fundamental (R165): CanCreateTask's limit check is only advisory — +// it is also called by RegisterTask without starting the task — so a +// check-then-act TOCTOU meant two concurrently starting tasks could both +// read current < limit and then both start, letting live tasks exceed +// maxConcurrent and defeating the DoS backstop. OnTaskStart is the +// single commit point, so it enforces maxConcurrent here via a CAS loop: +// only maxConcurrent tasks can hold slots at once. BackgroundTask.Start +// aborts (without launching run) when this returns false. +func (cb *TaskCircuitBreaker) OnTaskStart(taskName string) bool { + max := atomic.LoadInt32(&cb.maxConcurrent) + for { + current := atomic.LoadInt32(&cb.concurrentTasks) + if current >= max { + if cb.logger != nil { + cb.logger.Debug("Concurrent task limit reached, refusing slot for: %s", taskName) + } + return false + } + if atomic.CompareAndSwapInt32(&cb.concurrentTasks, current, current+1) { + break + } + } cb.tasksMu.Lock() cb.activeTasks[taskName] = struct{}{} cb.tasksMu.Unlock() @@ -358,6 +408,7 @@ func (cb *TaskCircuitBreaker) OnTaskStart(taskName string) { cb.logger.Debug("Task started, concurrent count: %d, task: %s", atomic.LoadInt32(&cb.concurrentTasks), taskName) } + return true } // OnTaskComplete records a task completing execution @@ -751,9 +802,14 @@ func (mm *TaskMemoryMonitor) checkForMemoryIssues(stats TaskMemoryStats) { historyLen := len(mm.statsHistory) if historyLen >= 2 { prev := mm.statsHistory[historyLen-2] - heapGrowth := float64(stats.HeapAlloc) / float64(prev.HeapAlloc) - if heapGrowth > 2.0 && stats.NumGC == prev.NumGC { - mm.logger.Infof("Potential memory leak: heap grew %.2fx without GC", heapGrowth) + // Guard against div-by-zero (prev.HeapAlloc == 0) which would make + // heapGrowth == +Inf and fire a spurious "memory leak" alert for a + // first-ever measurement (R103). + if prev.HeapAlloc > 0 { + heapGrowth := float64(stats.HeapAlloc) / float64(prev.HeapAlloc) + if heapGrowth > 2.0 && stats.NumGC == prev.NumGC { + mm.logger.Infof("Potential memory leak: heap grew %.2fx without GC", heapGrowth) + } } } mm.mu.RUnlock() diff --git a/error_recovery.go b/error_recovery.go index 832b61b..33c392a 100644 --- a/error_recovery.go +++ b/error_recovery.go @@ -243,6 +243,14 @@ func (cb *CircuitBreaker) ExecuteWithContext(ctx context.Context, fn func() erro cb.RecordRequest() if !cb.allowRequest() { + // A request rejected while the circuit is open is an admission + // outcome, not a never-started request: count it as a failure so + // GetBaseMetrics' success_rate (successes/total_requests) and + // total_failures reflect actual admission, matching the + // internal/recovery circuit breaker. Previously open-rejects only + // incremented total_requests, understating failures and + // deflating the reported success rate (R180). + cb.RecordFailure() return fmt.Errorf("circuit breaker is open") } @@ -362,7 +370,24 @@ func (cb *CircuitBreaker) Reset() { // IsAvailable returns whether the circuit breaker is currently allowing requests. // This provides a quick way to check if the service is available. func (cb *CircuitBreaker) IsAvailable() bool { - return cb.allowRequest() + cb.mutex.Lock() + defer cb.mutex.Unlock() + + switch cb.state { + case CircuitBreakerClosed: + return true + case CircuitBreakerHalfOpen: + return true + case CircuitBreakerOpen: + // Read-only availability probe: report that a request would be + // admitted only after the open timeout has elapsed, but do NOT + // transition to half-open — IsAvailable must not take a permit or + // mutate circuit state. Real traffic admission is gated by + // allowRequest (triggerRequest) which performs the transition. + return time.Since(cb.lastFailureTime) > cb.timeout + default: + return false + } } // GetMetrics returns comprehensive metrics about the circuit breaker. @@ -607,6 +632,18 @@ func (re *RetryExecutor) isRetryableError(err error) bool { return false } + // A terminal HTTP error (a real status that is neither 5xx nor 429) + // must be classified by its status BEFORE any message-substring scan. + // Otherwise a 4xx whose provider message merely contains a + // retryable-looking word — e.g. "certificate"/"tls" (isCertificateError), + // "EOF" (isEOFError), or a generic "timeout" — would be reclassified + // retryable and retried to MaxAttempts on a permanent error, repeating + // the failing request (R123, R157). + var statusHTTP *HTTPError + if errors.As(err, &statusHTTP) && statusHTTP.StatusCode != 0 && statusHTTP.StatusCode < 500 && statusHTTP.StatusCode != 429 { + return false + } + // Check for Traefik default certificate error (startup race condition) // See: https://github.com/lukaszraczylo/traefikoidc/issues/90 if isTraefikDefaultCertError(err) { @@ -631,7 +668,8 @@ func (re *RetryExecutor) isRetryableError(err error) bool { } } - if netErr, ok := err.(net.Error); ok { + var netErr net.Error + if errors.As(err, &netErr) { if netErr.Timeout() { return true } @@ -652,8 +690,18 @@ func (re *RetryExecutor) isRetryableError(err error) bool { } } - if httpErr, ok := err.(*HTTPError); ok { - return httpErr.StatusCode >= 500 || httpErr.StatusCode == 429 + // errors.As so retry classification survives error wrapping: a transient + // or 5xx/429 error returned as fmt.Errorf("...: %w", err) from an + // upstream layer must still be retried. Direct type assertions on the + // top-level error silently misclassified wrapped transient errors as + // permanent, giving up before any retry (R111). + var httpErr *HTTPError + if errors.As(err, &httpErr) { + status := httpErr.StatusCode + // A 0 StatusCode means the producer used HTTPError purely as a + // message carrier (no real HTTP status). Treat it as unknown + // (non-retryable) rather than spuriously matching 5xx (R118). + return status != 0 && (status >= 500 || status == 429) } return false @@ -880,11 +928,23 @@ var gdInstances = struct { set map[*GracefulDegradation]struct{} }{set: make(map[*GracefulDegradation]struct{})} +// globalHealthCheckMu serializes the global pass. globalPerformHealthChecks +// runs from more than one goroutine — the shared singleton task's run loop +// AND the once-guarded StartBackgroundTask path both drive it — so without +// this, any health check registered via the public RegisterHealthCheck +// API would be invoked concurrently (a latent production race, and the +// source of an intermittent full-suite race in tests). Serializing the +// whole pass keeps registered health checks single-threaded (R189). +var globalHealthCheckMu sync.Mutex + // globalPerformHealthChecks runs the health check pass for every live // GracefulDegradation instance. The shared singleton task calls this, so a // health check registered on ANY instance is exercised, and closing one // instance does not stall recovery for the others. func globalPerformHealthChecks() { + globalHealthCheckMu.Lock() + defer globalHealthCheckMu.Unlock() + gdInstances.RLock() instances := make([]*GracefulDegradation, 0, len(gdInstances.set)) for gd := range gdInstances.set { @@ -1201,8 +1261,14 @@ func (erm *ErrorRecoveryManager) GetCircuitBreaker(serviceName string) *CircuitB func (erm *ErrorRecoveryManager) ExecuteWithRecovery(ctx context.Context, serviceName string, fn func() error) error { cb := erm.GetCircuitBreaker(serviceName) - return erm.retryExecutor.Execute(ctx, func() error { - return cb.Execute(fn) + // Retry INNER, circuit OUTER (matching TokenResilienceManager and the + // documented R106 compositing rule). Wrapping cb around the whole retry + // run means one logical call counts as a single circuit failure even + // when it was retried internally; the previous nesting (cb inside the + // retry loop) recorded each retry attempt as a separate failure and + // could open the circuit after only one flaky-but-retryable call (R144). + return cb.ExecuteWithContext(ctx, func() error { + return erm.retryExecutor.Execute(ctx, fn) }) } diff --git a/goroutine_manager.go b/goroutine_manager.go index 25ef81a..8322917 100644 --- a/goroutine_manager.go +++ b/goroutine_manager.go @@ -63,6 +63,14 @@ func (m *GoroutineManager) StartGoroutine(name string, fn func(context.Context)) m.wg.Done() m.mu.Lock() managedGoroutine.running = false + // Prune the map entry once this goroutine has finished so that + // finished goroutines don't accumulate unboundedly (R157). Only + // delete when the current entry still IS the finishing + // goroutine: a concurrent Start with the same name may have + // already replaced it with a live one. + if cur, ok := m.goroutines[goroutineName]; ok && cur == managedGoroutine { + delete(m.goroutines, goroutineName) + } m.mu.Unlock() // Recover from panics diff --git a/http_client_pool.go b/http_client_pool.go index 972c4da..dee333c 100644 --- a/http_client_pool.go +++ b/http_client_pool.go @@ -81,10 +81,21 @@ func (p *SharedTransportPool) GetOrCreateTransport(config HTTPClientConfig) *htt return shared.transport } } - // No TLS-compatible transport available; return nil so the caller falls - // back to a default, certificate-verifying transport rather than one - // with a different (possibly verification-disabled) trust store. - return nil + // No TLS-compatible transport available. Returning nil would make the + // caller fall back to http.DefaultTransport, silently dropping the + // configured RootCAs / InsecureSkipVerify — custom-CA TLS would + // break and self-signed skip-verify would be lost (R95). Build and + // return the correctly-configured transport instead; honoring TLS + // settings takes precedence over the soft client cap. + t := newSharedTransport(config) + p.transports[p.configKey(config)] = &sharedTransport{ + transport: t, + refCount: 1, + lastUsed: time.Now(), + tlsKey: want, + } + atomic.AddInt32(&p.clientCount, 1) + return t } p.mu.Lock() @@ -101,8 +112,26 @@ func (p *SharedTransportPool) GetOrCreateTransport(config HTTPClientConfig) *htt // Increment client count atomic.AddInt32(&p.clientCount, 1) - // Create new transport with conservative limits - transport := &http.Transport{ + transport := newSharedTransport(config) + + p.transports[key] = &sharedTransport{ + transport: transport, + refCount: 1, + lastUsed: time.Now(), + tlsKey: tlsConfigKey(config), + } + + return transport +} + +// newSharedTransport builds a properly-configured *http.Transport from the +// given config, enforcing TLS 1.2+ and secure cipher suites. Shared by the +// normal acquisition path and the at-cap fallback so the TLS settings are +// NEVER silently dropped (the at-cap fallback previously returned nil, +// making the caller fall back to http.DefaultTransport and lose the +// configured RootCAs / InsecureSkipVerify — R95). +func newSharedTransport(config HTTPClientConfig) *http.Transport { + return &http.Transport{ Proxy: http.ProxyFromEnvironment, DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) { dialer := &net.Dialer{ @@ -138,15 +167,6 @@ func (p *SharedTransportPool) GetOrCreateTransport(config HTTPClientConfig) *htt WriteBufferSize: config.WriteBufferSize, ReadBufferSize: config.ReadBufferSize, } - - p.transports[key] = &sharedTransport{ - transport: transport, - refCount: 1, - lastUsed: time.Now(), - tlsKey: tlsConfigKey(config), - } - - return transport } // ReleaseTransport decrements the reference count for a transport @@ -270,20 +290,31 @@ func tlsConfigKey(config HTTPClientConfig) string { return fmt.Sprintf("%p|%s", config.RootCAs, skip) } -// Cleanup closes all transports and stops the cleanup goroutine +// Cleanup closes all transports, resets the pool, and restarts the cleanup +// goroutine so the (singleton) pool remains fully usable afterwards. Prior +// to R125, Cleanup canceled the cleanup goroutine but never restarted it +// and left clientCount unreset: a reused singleton then had a permanently +// dead cleaner (idle conns never pruned) and a stale count (the soft +// maxClients cap keyed off it was wrong after reuse). func (p *SharedTransportPool) Cleanup() { p.mu.Lock() - defer p.mu.Unlock() - - // Stop the cleanup goroutine - if p.cancel != nil { - p.cancel() - } - for _, shared := range p.transports { - shared.transport.CloseIdleConnections() + if shared != nil && shared.transport != nil { + shared.transport.CloseIdleConnections() + } } p.transports = make(map[string]*sharedTransport) + atomic.StoreInt32(&p.clientCount, 0) + oldCancel := p.cancel + ctx, cancel := context.WithCancel(context.Background()) + p.ctx = ctx + p.cancel = cancel + p.mu.Unlock() + + if oldCancel != nil { + oldCancel() + } + go p.cleanupIdleTransports(ctx) } // CreatePooledHTTPClient creates an HTTP client using the shared transport pool diff --git a/internal/cache/backends/hybrid.go b/internal/cache/backends/hybrid.go index eccd998..865a9ee 100644 --- a/internal/cache/backends/hybrid.go +++ b/internal/cache/backends/hybrid.go @@ -37,7 +37,6 @@ type HybridBackend struct { // asyncWriteItem represents an async write operation type asyncWriteItem struct { - ctx context.Context key string value []byte ttl time.Duration @@ -195,7 +194,6 @@ func (h *HybridBackend) Set(ctx context.Context, key string, value []byte, ttl t key: key, value: value, ttl: ttl, - ctx: ctx, }: h.logger.Debugf("Queued async write to L2 for key: %s", redactKey(key)) default: @@ -265,14 +263,23 @@ func (h *HybridBackend) Delete(ctx context.Context, key string) (bool, error) { deleted = true } - // Delete from L2 if not in fallback mode - if !h.fallbackMode.Load() { - if d, err := h.secondary.Delete(ctx, key); err != nil { - h.logger.Debugf("Failed to delete from L2 cache: %v", err) - h.recordL2Error() - } else if d { - deleted = true - } + if h.fallbackMode.Load() { + return deleted, nil + } + + // Drain in-flight async L2 writes before deleting L2. A Set's async + // write could otherwise land AFTER this DEL, resurrecting the just- + // deleted key with its stale value (L1/L2 divergence). Waiting for + // the async worker (which tracks queued + in-flight non-critical + // writes) ensures no pre-existing async Set lands post-DEL (R124). + h.asyncWg.Wait() + + // Delete from L2 + if d, err := h.secondary.Delete(ctx, key); err != nil { + h.logger.Debugf("Failed to delete from L2 cache: %v", err) + h.recordL2Error() + } else if d { + deleted = true } return deleted, nil @@ -390,12 +397,14 @@ func (h *HybridBackend) Ping(ctx context.Context) error { // Close shuts down the hybrid backend func (h *HybridBackend) Close() error { - // Cancel context to stop workers + // Cancel context to stop workers. h.cancel() - // Close async write channel - close(h.asyncWriteBuffer) - close(h.l1BackfillBuffer) + // Do NOT close asyncWriteBuffer / l1BackfillBuffer: the workers key off + // h.ctx.Done() and exit on cancel, and closing these channels creates a + // race where a concurrent Set's select-send (line ~193) hits a closed + // channel and panics ("send on closed channel"). Buffered items a late + // Set queues after the workers have exited are harmless at shutdown. // Wait for workers to finish with timeout done := make(chan struct{}) @@ -539,7 +548,6 @@ func (h *HybridBackend) SetMany(ctx context.Context, items map[string][]byte, tt key: key, value: value, ttl: ttl, - ctx: ctx, }: // Queued default: @@ -593,7 +601,7 @@ func (h *HybridBackend) l1BackfillWorker() { if !ok { return } - if h.keyStillInSecondary(item.key) { + if h.keyStillInSecondary(item.key) && !h.keyAlreadyInPrimary(item.key) { writeCtx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) if err := h.primary.Set(writeCtx, item.key, item.value, item.ttl); err != nil { h.logger.Debugf("Failed to populate L1 cache from L2 for key %s: %v", redactKey(item.key), err) @@ -620,6 +628,19 @@ func (h *HybridBackend) keyStillInSecondary(key string) bool { return err != nil || exists } +// keyAlreadyInPrimary reports whether the key already exists in the L1 +// cache. The backfill worker writes an L2-resolved value into L1; if a +// concurrent Set already placed a fresher value there, writing over it +// would clobber the newer value with the stale L2 snapshot (R124). Skip +// the backfill when L1 already has the key — L1 is only a cache and the +// next L2 hit re-arms the write-through. +func (h *HybridBackend) keyAlreadyInPrimary(key string) bool { + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + exists, err := h.primary.Exists(ctx, key) + return err == nil && exists +} + // asyncWriteWorker processes asynchronous writes to L2 func (h *HybridBackend) asyncWriteWorker() { defer h.wg.Done() @@ -658,7 +679,11 @@ func (h *HybridBackend) asyncWriteWorker() { // Clear stall us out so no buffered write can land after a // clear-then-wipe. h.asyncFlushMu.Lock() - writeCtx, cancel := context.WithTimeout(item.ctx, 500*time.Millisecond) + // Use a detached context, never the originating request's ctx: + // this is a background write that typically runs after the + // request has returned, so the request ctx would be canceled + // and the L2 write silently dropped (R94). + writeCtx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond) if err := h.secondary.Set(writeCtx, item.key, item.value, item.ttl); err != nil { h.errors.Add(1) h.logger.Debugf("Async write to L2 failed for key %s: %v", redactKey(item.key), err) diff --git a/internal/cache/backends/hybrid_r94_test.go b/internal/cache/backends/hybrid_r94_test.go new file mode 100644 index 0000000..1b5f1a2 --- /dev/null +++ b/internal/cache/backends/hybrid_r94_test.go @@ -0,0 +1,75 @@ +package backends + +import ( + "context" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// r94CtxProbeBackend records whether any operation was handed a canceled +// context. Set is the only method the async L2 write path exercises. +type r94CtxProbeBackend struct { + setSeen *atomic.Bool + setHadErr *atomic.Bool +} + +func (b *r94CtxProbeBackend) Set(ctx context.Context, key string, value []byte, ttl time.Duration) error { + b.setSeen.Store(true) + if ctx.Err() != nil { + b.setHadErr.Store(true) + } + return nil +} +func (b *r94CtxProbeBackend) Get(ctx context.Context, key string) ([]byte, time.Duration, bool, error) { + return nil, 0, false, nil +} +func (b *r94CtxProbeBackend) Delete(ctx context.Context, key string) (bool, error) { return false, nil } +func (b *r94CtxProbeBackend) Exists(ctx context.Context, key string) (bool, error) { return false, nil } +func (b *r94CtxProbeBackend) Clear(ctx context.Context) error { return nil } +func (b *r94CtxProbeBackend) GetStats() map[string]interface{} { return nil } +func (b *r94CtxProbeBackend) Close() error { return nil } +func (b *r94CtxProbeBackend) Ping(ctx context.Context) error { return nil } + +// TestHybridBackend_AsyncWriteUsesDetachedContext is an R94 regression: the +// async L2 write-through must be executed with a detached context, never the +// originating request's context. The L2 write runs after the request has +// returned; if it carried the (now-canceled) request context the write +// would be silently dropped, so non-critical keys never reach L2. Before +// the fix the worker used the request ctx and recorded a canceled context. +func TestHybridBackend_AsyncWriteUsesDetachedContext(t *testing.T) { + primary := &r94CtxProbeBackend{setSeen: &atomic.Bool{}, setHadErr: &atomic.Bool{}} + secondary := &r94CtxProbeBackend{setSeen: &atomic.Bool{}, setHadErr: &atomic.Bool{}} + + hybrid, err := NewHybridBackend(&HybridConfig{ + Primary: primary, + Secondary: secondary, + SyncWriteCacheTypes: map[string]bool{ + // empty set: no type is sync, so writes are async + }, + }) + require.NoError(t, err) + defer hybrid.Close() + + reqCtx, cancel := context.WithCancel(context.Background()) + err = hybrid.Set(reqCtx, "general:key1", []byte("test-value"), time.Minute) + require.NoError(t, err) + + // The request has returned: cancel its context before the async worker + // gets a chance to run the L2 write. + cancel() + + // Wait for the async L2 write to be attempted. + deadline := time.Now().Add(2 * time.Second) + for !secondary.setSeen.Load() { + if time.Now().After(deadline) { + t.Fatal("async L2 write was never attempted") + } + time.Sleep(5 * time.Millisecond) + } + + require.False(t, secondary.setHadErr.Load(), + "async L2 write must use a detached (non-canceled) context") +} diff --git a/internal/cache/backends/memory.go b/internal/cache/backends/memory.go index 61c0547..57f8a46 100644 --- a/internal/cache/backends/memory.go +++ b/internal/cache/backends/memory.go @@ -202,6 +202,14 @@ func (m *MemoryCacheBackend) Set(ctx context.Context, key string, value interfac return ErrBackendUnavailable } + // A NEGATIVE TTL means "already expired" (the stack's convention for + // a value whose validity has passed — see redis.go). Previously ttl<=0 + // fell through to a permanent entry, diverging from redis and + // universal_cache. Zero remains the documented "no expiry" contract. + if ttl < 0 { + return nil + } + start := time.Now() defer func() { duration := time.Since(start).Nanoseconds() @@ -212,15 +220,19 @@ func (m *MemoryCacheBackend) Set(ctx context.Context, key string, value interfac // Calculate item size itemSize := int64(len(key)) + estimateValueSize(value) - // Enforce global limits before adding new item - m.enforceGlobalLimits(itemSize) + // Snapshot replacement state so the global limit enforcer accounts for + // key replacement: refreshing an existing key neither grows the count + // nor adds its full size, so at global capacity it must not evict an + // unrelated live entry (mirrors cacheShard.set, R177). + shard := m.getShard(key) + oldSize, exists := shard.peekSize(key) + m.enforceGlobalLimits(itemSize, oldSize, exists) var expiresAt time.Time if ttl > 0 { expiresAt = time.Now().Add(ttl) } - shard := m.getShard(key) shard.set(key, value, expiresAt, itemSize) m.sets.Add(1) @@ -228,14 +240,30 @@ func (m *MemoryCacheBackend) Set(ctx context.Context, key string, value interfac } // enforceGlobalLimits ensures global size and memory limits are respected -// by evicting from shards when necessary -func (m *MemoryCacheBackend) enforceGlobalLimits(newItemSize int64) { - // Check and enforce size limit +// by evicting from shards when necessary. +// +// It is replacement-aware: an existing key being rewritten (exists=oldSize) +// is count-neutral (global cap is on the number of entries) and grows +// memory by only the net delta, mirroring cacheShard.set. At global +// capacity a pure key replacement must therefore not evict an unrelated +// live entry (R177). +func (m *MemoryCacheBackend) enforceGlobalLimits(newItemSize, oldSize int64, exists bool) { + // Count (entry-size) budget only grows for genuinely new keys. + sizeTrigger := m.maxSize > 0 && !exists + // Memory budget grows by the net delta for a replacement. + netMemory := newItemSize + if exists { + netMemory = newItemSize - oldSize + if netMemory < 0 { + netMemory = 0 + } + } + for { totalSize, totalMemory := m.getGlobalStats() - needsSizeEviction := m.maxSize > 0 && totalSize >= m.maxSize - needsMemoryEviction := m.maxMemory > 0 && totalMemory+newItemSize > m.maxMemory + needsSizeEviction := sizeTrigger && totalSize >= m.maxSize + needsMemoryEviction := m.maxMemory > 0 && totalMemory+netMemory > m.maxMemory if !needsSizeEviction && !needsMemoryEviction { break diff --git a/internal/cache/backends/memory_shard.go b/internal/cache/backends/memory_shard.go index a1e3181..b791ae6 100644 --- a/internal/cache/backends/memory_shard.go +++ b/internal/cache/backends/memory_shard.go @@ -30,6 +30,19 @@ func newCacheShard(maxSize, maxMemory int64) *cacheShard { // get retrieves a value from this shard // Returns: value, exists, expired +// peekSize returns the size of the existing entry for key and whether it +// exists, without touching LRU state. Used by the global limit enforcer +// to decide whether a Set is a replacement (count-neutral) or a new item. +func (s *cacheShard) peekSize(key string) (int64, bool) { + s.mu.RLock() + defer s.mu.RUnlock() + item, exists := s.items[key] + if !exists { + return 0, false + } + return item.size, true +} + func (s *cacheShard) get(key string) (interface{}, bool, bool) { s.mu.RLock() item, exists := s.items[key] @@ -72,15 +85,27 @@ func (s *cacheShard) set(key string, value interface{}, expiresAt time.Time, siz s.mu.Lock() defer s.mu.Unlock() - // Check if we need to evict items - if s.maxSize > 0 && s.size >= s.maxSize { - s.evictLRULocked() - } - if s.maxMemory > 0 && s.memoryUsed+size > s.maxMemory { - s.evictLRULocked() + // Snapshot the existing entry so the eviction decisions account for + // replacement: refreshing an existing key neither grows the count nor + // adds its full size, so at capacity it must not evict an unrelated + // LRU entry (R139 F2). + oldItem, exists := s.items[key] + var oldSize int64 + if exists { + oldSize = oldItem.size } - // Remove old item if exists + if s.maxSize > 0 && !exists && s.size >= s.maxSize { + s.evictLRULocked() + } + if s.maxMemory > 0 { + if net := size - oldSize; net > 0 && s.memoryUsed+net > s.maxMemory { + s.evictLRULocked() + } + } + + // Remove old item if it still exists (it may have been the LRU victim + // of an eviction above; re-check so we don't double-account). if oldItem, exists := s.items[key]; exists { s.memoryUsed -= oldItem.size if elem, ok := oldItem.element.(*list.Element); ok && elem != nil { diff --git a/internal/cache/backends/redis.go b/internal/cache/backends/redis.go index e906efc..420884e 100644 --- a/internal/cache/backends/redis.go +++ b/internal/cache/backends/redis.go @@ -95,7 +95,15 @@ func (r *RedisBackend) Set(ctx context.Context, key string, value []byte, ttl ti return ErrBackendClosed } + // A NEGATIVE TTL means "already expired" (the stack's convention for a + // value whose validity has passed). Previously ttl<=0 fell through to a + // bare SET with no expiry, silently making the entry permanent. Zero here + // is the documented "no expiry" contract (see SetManyNoTTL), so only + // strictly-negative TTLs are skipped. prefixedKey := r.prefixKey(key) + if ttl < 0 { + return nil + } // Execute with retry logic return r.executeWithRetry(ctx, func(conn *RedisConn) error { @@ -450,6 +458,13 @@ func (r *RedisBackend) SetMany(ctx context.Context, items map[string][]byte, ttl return nil } + if ttl < 0 { + // Already-expired TTL: nothing to persist (see Set's guard). Avoids + // creating permanent entries for past-dated values via the multi path. + // Zero retains the "no expiry" contract (see SetManyNoTTL). + return nil + } + // For single items, use regular Set if len(items) == 1 { for key, value := range items { diff --git a/internal/cache/backends/redis_pool.go b/internal/cache/backends/redis_pool.go index 18ba06e..da89316 100644 --- a/internal/cache/backends/redis_pool.go +++ b/internal/cache/backends/redis_pool.go @@ -67,6 +67,22 @@ func NewConnectionPool(config *PoolConfig) (*ConnectionPool, error) { return pool, nil } +// reserveConnectionSlot atomically reserves capacity for one new connection, +// returning false when the pool is already at MaxConnections. The +// reservation is released via totalConns.Add(-1) if the dial fails. +func (p *ConnectionPool) reserveConnectionSlot() bool { + max := int32(p.config.MaxConnections) // nolint:gosec // MaxConnections is operator-bounded well below int32 max + for { + cur := p.totalConns.Load() + if cur >= max { + return false + } + if p.totalConns.CompareAndSwap(cur, cur+1) { + return true + } + } +} + // Get retrieves a connection from the pool or creates a new one func (p *ConnectionPool) Get(ctx context.Context) (*RedisConn, error) { if p.closed.Load() { @@ -102,11 +118,15 @@ func (p *ConnectionPool) Get(ctx context.Context) (*RedisConn, error) { return nil, ctx.Err() default: - // No available connection, create new one if under limit - // #nosec G115 -- MaxConnections is a small config value that fits in int32 - if p.totalConns.Load() < int32(p.config.MaxConnections) { + // No available connection, create new one if under limit. + // Reserve capacity atomically before dialing: checking + // totalConns then dialing is not atomic, so concurrent Gets + // could each pass the check and spike the live socket count + // far above MaxConnections (FD / Redis maxclients exhaustion). + if p.reserveConnectionSlot() { conn, err = p.createConnection(ctx) if err != nil { + p.totalConns.Add(-1) // release the reserved slot // If this is the last attempt, return error if attempt == maxAttempts-1 { return nil, err @@ -116,7 +136,6 @@ func (p *ConnectionPool) Get(ctx context.Context) (*RedisConn, error) { continue } p.activeConns.Add(1) - p.totalConns.Add(1) return conn, nil } @@ -335,7 +354,9 @@ func (c *RedisConn) Do(command string, args ...string) (interface{}, error) { resp, err := reader.ReadResponse() reader.Release() // Return to pool immediately after use if err != nil { - if !errors.Is(err, ErrNilResponse) { + // A nil response or a Redis command error reply leaves the + // connection healthy — only an IO/parse error closes it. + if !errors.Is(err, ErrNilResponse) && !errors.Is(err, ErrCommandReply) { c.closed.Store(true) } return nil, err @@ -470,6 +491,14 @@ func (p *Pipeline) Execute() ([]interface{}, error) { responses[i] = nil continue } + if errors.Is(err, ErrCommandReply) { + // A Redis command error reply: the connection is healthy + // and the remaining replies are still readable. Store the + // error value so per-command validation upstream can + // report it without aborting the batch. + responses[i] = err + continue + } p.conn.closed.Store(true) return responses[:i], fmt.Errorf("pipeline read error at command %d: %w", i, err) } diff --git a/internal/cache/backends/resp.go b/internal/cache/backends/resp.go index 3b7fde7..85bc6be 100644 --- a/internal/cache/backends/resp.go +++ b/internal/cache/backends/resp.go @@ -20,6 +20,11 @@ import ( var ( ErrInvalidRESP = errors.New("invalid RESP response") ErrNilResponse = errors.New("nil response") + // ErrCommandReply marks a RESP '-' (Redis command error) reply, e.g. + // WRONGTYPE or OOM. It is a valid protocol response — the connection + // is healthy and any remaining pipelined replies are still readable — + // as opposed to an IO/parse error which means the connection is dead. + ErrCommandReply = errors.New("redis command error reply") ) // RESPWriter writes RESP protocol messages @@ -110,7 +115,9 @@ func (r *RESPReader) readError() error { if err != nil { return err } - return errors.New(line) + // Wrap so callers can distinguish a Redis command error reply (healthy + // connection) from an IO/parse error (dead connection). + return fmt.Errorf("%w: %s", ErrCommandReply, line) } // readInteger reads an integer (:1000\r\n) diff --git a/internal/cache/backends/resp_test.go b/internal/cache/backends/resp_test.go index 3b4ae6b..88a5573 100644 --- a/internal/cache/backends/resp_test.go +++ b/internal/cache/backends/resp_test.go @@ -142,7 +142,12 @@ func TestRESPReader_ReadError(t *testing.T) { _, err := reader.ReadResponse() require.Error(t, err) - assert.Equal(t, tt.expectedError, err.Error()) + // R177: a '-' reply is now wrapped with ErrCommandReply so + // callers can tell a Redis command error (healthy connection) + // apart from an IO/parse error. The command's message text is + // preserved verbatim inside the wrapped error. + assert.True(t, errors.Is(err, ErrCommandReply), "want ErrCommandReply, got %v", err) + assert.Contains(t, err.Error(), tt.expectedError) }) } } diff --git a/internal/cache/backends/review_r110_test.go b/internal/cache/backends/review_r110_test.go new file mode 100644 index 0000000..81c9e23 --- /dev/null +++ b/internal/cache/backends/review_r110_test.go @@ -0,0 +1,54 @@ +package backends + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// TestConnectionPool_CapHeldUnderBurst regresses the connection-pool cap +// being non-atomic: ConnectionPool.Get checked `totalConns < MaxConnections` +// and THEN dialed a new socket, so concurrent Gets could each pass the check +// and spike the live connection count far above MaxConnections (FD / Redis +// maxclients exhaustion under burst load). The cap must now be held +// atomically so at most MaxConnections sockets are ever created at once. +func TestConnectionPool_CapHeldUnderBurst(t *testing.T) { + mr := NewMiniredisServer(t) + + maxConns := 3 + pool, err := NewConnectionPool(&PoolConfig{ + Address: mr.GetAddr(), + MaxConnections: maxConns, + ConnectTimeout: 80 * time.Millisecond, + ReadTimeout: 50 * time.Millisecond, + WriteTimeout: 50 * time.Millisecond, + }) + require.NoError(t, err) + defer pool.Close() + + ctx := context.Background() + const workers = 50 + + var wg sync.WaitGroup + for i := 0; i < workers; i++ { + wg.Add(1) + go func() { + defer wg.Done() + conn, _ := pool.Get(ctx) + if conn != nil { + // Hold the connection (simulate an in-flight request) so the + // pool stays under pressure and the count is stable. + _ = conn + } + }() + } + wg.Wait() + + stats := pool.Stats() + total := stats["total_connections"].(int32) + require.LessOrEqual(t, total, int32(maxConns), + "burst Gets must not create more than MaxConnections sockets (got %d)", total) +} diff --git a/internal/cache/backends/review_r115_test.go b/internal/cache/backends/review_r115_test.go new file mode 100644 index 0000000..f63f2ec --- /dev/null +++ b/internal/cache/backends/review_r115_test.go @@ -0,0 +1,36 @@ +package backends + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// TestRedisBackend_SetNegativeTTL_NotPersistent guards the R115 negative-TTL +// fix. The cache stack's convention is that a non-positive TTL means "already +// expired". Previously RedisBackend.Set/SetMany fell through to a bare SET +// when ttl<=0, writing an entry with NO expiry — i.e. an expired-dated value +// became permanent. Now expired TTLs write nothing. +func TestRedisBackend_SetNegativeTTL_NotPersistent(t *testing.T) { + _, backend := setupTestRedis(t) + ctx := context.Background() + + // Single Set with an expired TTL must not create an entry. + require.NoError(t, backend.Set(ctx, "neg", []byte("v"), -time.Second)) + _, _, exists, err := backend.Get(ctx, "neg") + require.NoError(t, err) + require.False(t, exists, "negative-TTL Set must not create a persistent entry") + + // SetMany with multiple items and an expired TTL must not create entries. + require.NoError(t, backend.SetMany(ctx, map[string][]byte{ + "a": []byte("1"), + "b": []byte("2"), + }, -time.Second)) + for _, k := range []string{"a", "b"} { + _, _, exists, err := backend.Get(ctx, k) + require.NoError(t, err) + require.False(t, exists, "negative-TTL SetMany must not create a persistent entry for %q", k) + } +} diff --git a/internal/cache/backends/review_r124_test.go b/internal/cache/backends/review_r124_test.go new file mode 100644 index 0000000..ffb70d0 --- /dev/null +++ b/internal/cache/backends/review_r124_test.go @@ -0,0 +1,241 @@ +package backends + +import ( + "context" + "sync" + "sync/atomic" + "testing" + "time" +) + +// blockingWriteBackend is a cache backend whose Set blocks until released, +// used to hold an async L2 write in-flight while Delete runs. +type blockingWriteBackend struct { + mu sync.Mutex + data map[string]string + setStarted chan struct{} + release chan struct{} + written chan struct{} + once sync.Once +} + +func (b *blockingWriteBackend) Set(ctx context.Context, key string, value []byte, ttl time.Duration) error { + select { + case <-b.setStarted: + default: + close(b.setStarted) + } + select { + case <-b.release: + case <-b.written: // never + case <-ctx.Done(): + return nil + } + b.mu.Lock() + b.data[key] = string(value) + b.mu.Unlock() + b.once.Do(func() { close(b.written) }) + return nil +} + +func (b *blockingWriteBackend) Get(ctx context.Context, key string) ([]byte, time.Duration, bool, error) { + b.mu.Lock() + defer b.mu.Unlock() + v, ok := b.data[key] + if !ok { + return nil, 0, false, nil + } + return []byte(v), time.Minute, true, nil +} + +func (b *blockingWriteBackend) Delete(ctx context.Context, key string) (bool, error) { + b.mu.Lock() + defer b.mu.Unlock() + _, ok := b.data[key] + delete(b.data, key) + return ok, nil +} + +func (b *blockingWriteBackend) Exists(ctx context.Context, key string) (bool, error) { + b.mu.Lock() + defer b.mu.Unlock() + _, ok := b.data[key] + return ok, nil +} + +func (b *blockingWriteBackend) Clear(ctx context.Context) error { + b.mu.Lock() + defer b.mu.Unlock() + b.data = map[string]string{} + return nil +} + +func (b *blockingWriteBackend) GetStats() map[string]interface{} { return nil } +func (b *blockingWriteBackend) Close() error { return nil } +func (b *blockingWriteBackend) Ping(ctx context.Context) error { return nil } + +// TestHybrid_DeleteWaitsForAsyncWrite guards the R124 fix to hybrid.go +// Delete: an in-flight async L2 write (from a non-critical Set) could land +// after Delete's L2 DEL, resurrecting the just-deleted key with a stale +// value. Delete must wait for in-flight async writes before the L2 delete. +func TestHybrid_DeleteWaitsForAsyncWrite(t *testing.T) { + sec := &blockingWriteBackend{ + data: map[string]string{}, + setStarted: make(chan struct{}), + release: make(chan struct{}), + written: make(chan struct{}), + } + primary := newMockBackend() + + hybrid, err := NewHybridBackend(&HybridConfig{ + Primary: primary, + Secondary: sec, + AsyncBufferSize: 16, + }) + if err != nil { + t.Fatalf("hybrid: %v", err) + } + defer hybrid.Close() + + ctx := context.Background() + + // Non-critical Set -> async L2 write that blocks in-flight. + if err := hybrid.Set(ctx, "k", []byte("stale"), time.Minute); err != nil { + t.Fatalf("set: %v", err) + } + select { + case <-sec.setStarted: + case <-time.After(2 * time.Second): + t.Fatal("async L2 write never started") + } + + // Issue Delete while the async write is in-flight. + done := make(chan struct{}) + go func() { + _, _ = hybrid.Delete(ctx, "k") + close(done) + }() + + // OLD behavior: Delete returns immediately (no wait for async writes); + // it performs its L2 delete and closes done. NEW behavior: Delete + // blocks in asyncWg.Wait until the in-flight write finishes, so done + // does not fire until released. Distinguish the two before releasing. + oldImmediate := false + select { + case <-done: + oldImmediate = true + case <-time.After(50 * time.Millisecond): + // NEW: Delete is blocked waiting for the async write. 50ms is + // far longer than OLD's immediate (non-waiting) return needs, + // and far shorter than the worker's 500ms writeCtx, so the + // in-flight write reliably survives to be released. + } + // Release the in-flight L2 write (well within the worker's writeCtx + // timeout) so it actually lands. + close(sec.release) + + if oldImmediate { + // OLD: the worker write now lands AFTER Delete's L2 delete, + // resurrecting the key. + select { + case <-sec.written: + case <-time.After(2 * time.Second): + t.Fatal("async L2 write never landed") + } + } else { + // NEW: Delete waited for the write, then removed it from L2. + select { + case <-sec.written: + case <-time.After(2 * time.Second): + t.Fatal("async L2 write never landed") + } + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("Delete did not return") + } + } + + if _, _, exists, _ := sec.Get(ctx, "k"); exists { + t.Fatal("L2 must not contain key after Delete: the in-flight async Set resurrected it") + } +} + +// existsAlwaysBackend reports the key always present in L1 (so a backfill +// must be skipped) and counts Set calls; setCalled is closed once Set runs, +// letting the test observe a (OLD) backfill write deterministically. +type existsAlwaysBackend struct { + setCalls int32 + setCalled chan struct{} + once sync.Once +} + +func newExistsAlwaysBackend() *existsAlwaysBackend { + return &existsAlwaysBackend{setCalled: make(chan struct{})} +} + +func (b *existsAlwaysBackend) Set(ctx context.Context, key string, value []byte, ttl time.Duration) error { + atomic.AddInt32(&b.setCalls, 1) + b.once.Do(func() { close(b.setCalled) }) + return nil +} +func (b *existsAlwaysBackend) Get(ctx context.Context, key string) ([]byte, time.Duration, bool, error) { + return nil, 0, false, nil +} +func (b *existsAlwaysBackend) Delete(ctx context.Context, key string) (bool, error) { + return false, nil +} +func (b *existsAlwaysBackend) Exists(ctx context.Context, key string) (bool, error) { return true, nil } +func (b *existsAlwaysBackend) Clear(ctx context.Context) error { return nil } +func (b *existsAlwaysBackend) GetStats() map[string]interface{} { return nil } +func (b *existsAlwaysBackend) Close() error { return nil } +func (b *existsAlwaysBackend) Ping(ctx context.Context) error { return nil } + +// TestHybrid_BackfillDoesNotClobberFreshL1 guards the R124 fix to +// hybrid.go l1BackfillWorker: the backfill only checked L2 key +// existence, so a stale L2-captured value would overwrite a fresher L1 +// value written concurrently by Set. When L1 already holds the key, the +// backfill must be skipped. +func TestHybrid_BackfillDoesNotClobberFreshL1(t *testing.T) { + primary := newExistsAlwaysBackend() + secondary := newMockBackend() + ctx := context.Background() + if err := secondary.Set(ctx, "k", []byte("stale"), time.Minute); err != nil { + t.Fatalf("seed L2: %v", err) + } + + hybrid, err := NewHybridBackend(&HybridConfig{ + Primary: primary, + Secondary: secondary, + AsyncBufferSize: 16, + }) + if err != nil { + t.Fatalf("hybrid: %v", err) + } + defer hybrid.Close() + + // Queue a stale backfill for a key L1 already holds (fresher). + hybrid.queueL1Backfill("k", []byte("stale"), time.Minute) + + // Wait for the worker to drain the backfill buffer. + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if len(hybrid.l1BackfillBuffer) == 0 { + time.Sleep(20 * time.Millisecond) // let the worker act (write or skip) + break + } + time.Sleep(5 * time.Millisecond) + } + + // Deterministically observe an OLD backfill write: if the worker + // writes, setCalled closes and setCalls becomes 1. On NEW the + // worker skips, so nothing fires. + select { + case <-primary.setCalled: + case <-time.After(500 * time.Millisecond): + } + + if atomic.LoadInt32(&primary.setCalls) != 0 { + t.Fatal("backfill must not overwrite the fresher L1 value (primary already had the key)") + } +} diff --git a/internal/cache/backends/review_r125_test.go b/internal/cache/backends/review_r125_test.go new file mode 100644 index 0000000..8ee0ba5 --- /dev/null +++ b/internal/cache/backends/review_r125_test.go @@ -0,0 +1,28 @@ +package backends + +import ( + "context" + "errors" + "testing" + "time" +) + +// TestMemoryBackend_NegativeTTLNotStored guards the R125 fix to memory.go: +// a negative TTL must mean "already expired" (the stack's convention, per +// redis.go) and NOT store a permanent entry. Previously ttl<=0 fell +// through to a permanent entry, diverging from redis and universal_cache, +// so a value whose validity had passed was kept forever in the memory +// tier. +func TestMemoryBackend_NegativeTTLNotStored(t *testing.T) { + b := NewMemoryCacheBackend(1000, 0, 0) + defer b.Close() + + ctx := context.Background() + if err := b.Set(ctx, "k", []byte("v"), -1*time.Second); err != nil { + t.Fatalf("set with negative TTL: %v", err) + } + + if _, err := b.Get(ctx, "k"); !errors.Is(err, ErrCacheMiss) { + t.Fatalf("Get after negative-TTL Set = %v, want %v (negative TTL must be treated as already-expired, not stored)", err, ErrCacheMiss) + } +} diff --git a/internal/cache/backends/review_r139_f2_test.go b/internal/cache/backends/review_r139_f2_test.go new file mode 100644 index 0000000..2f88767 --- /dev/null +++ b/internal/cache/backends/review_r139_f2_test.go @@ -0,0 +1,28 @@ +package backends + +import ( + "testing" + "time" +) + +// TestCacheShard_SetRefreshExistingDoesNotEvictAtCapacity verifies the shard +// set path does not evict an unrelated LRU entry when replacing an existing +// key at capacity (count stays flat) (R139 F2). Mirrors the Cache.Set fix. +func TestCacheShard_SetRefreshExistingDoesNotEvictAtCapacity(t *testing.T) { + s := newCacheShard(2, 0) // maxSize=2, no memory bound + exp := time.Now().Add(time.Minute) + + s.set("a", "va", exp, 1) // becomes LRU + s.set("b", "vb", exp, 1) // MRU + + // Refresh the most-recent key while at capacity. On the buggy path the + // LRU sibling `a` is evicted; on the fixed path the replacement is free. + s.set("b", "vb2", exp, 1) + + if _, exists, _ := s.get("a"); !exists { + t.Fatal("refreshing an existing key must not evict an unrelated live entry `a`") + } + if v, exists, _ := s.get("b"); !exists || v != "vb2" { + t.Fatalf("refreshed key `b` should hold the new value, got %v exists=%v", v, exists) + } +} diff --git a/internal/cache/backends/review_r141_test.go b/internal/cache/backends/review_r141_test.go new file mode 100644 index 0000000..852b4b7 --- /dev/null +++ b/internal/cache/backends/review_r141_test.go @@ -0,0 +1,81 @@ +package backends + +import ( + "context" + "sync" + "testing" + "time" +) + +// TestSingleflightCache_PanickingFetcherDoesNotDeadlock verifies that a +// fetcher which panics (routines in this codebase treat panics as a normal +// failure mode) does not wedge the singleflight entry: the leading caller +// yields an error and waiting callers are released instead of blocking on +// call.wg forever (R141). +func TestSingleflightCache_PanickingFetcherDoesNotDeadlock(t *testing.T) { + backend, err := NewMemoryBackend(DefaultConfig()) + requireNoErrorFatal(t, err) + defer backend.Close() + + cache := NewSingleflightCache(backend) + ctx := context.Background() + key := "panic-key" + + panicFetcher := func(ctx context.Context) ([]byte, time.Duration, error) { + panic("boom") + } + + var wg sync.WaitGroup + var doneA, doneB = make(chan struct{}), make(chan struct{}) + + wg.Add(1) + go func() { + defer wg.Done() + defer func() { _ = recover() }() // tolerate old code's out-of-band panic + _, _ = cache.GetOrFetch(ctx, key, panicFetcher) + close(doneA) + }() + + wg.Add(1) + go func() { + defer wg.Done() + _, _ = cache.GetOrFetch(ctx, key, panicFetcher) + close(doneB) + }() + + select { + case <-doneB: + case <-time.After(3 * time.Second): + t.Fatal("concurrent waiter hung on a panicking singleflight fetcher") + } + select { + case <-doneA: + case <-time.After(3 * time.Second): + t.Fatal("leading caller hung on its own panicking fetcher") + } + wg.Wait() + + // After the entry is cleaned up, a well-behaved fetcher for the same + // key must execute normally (proves the entry was not permanently + // wedged by the panic). + ran := false + good := func(ctx context.Context) ([]byte, time.Duration, error) { + ran = true + return []byte("ok"), time.Minute, nil + } + // Give the deferred cleanup a moment to release the key. + time.Sleep(200 * time.Millisecond) + if _, err := cache.GetOrFetch(ctx, key, good); err != nil { + t.Fatalf("well-behaved fetcher after panic returned error: %v", err) + } + if !ran { + t.Fatal("well-behaved fetcher after panic did not execute") + } +} + +func requireNoErrorFatal(t *testing.T, err error) { + t.Helper() + if err != nil { + t.Fatalf("unexpected error: %v", err) + } +} diff --git a/internal/cache/backends/review_r177_redis_test.go b/internal/cache/backends/review_r177_redis_test.go new file mode 100644 index 0000000..993fd1a --- /dev/null +++ b/internal/cache/backends/review_r177_redis_test.go @@ -0,0 +1,79 @@ +package backends + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// TestR177_RedisCommandErrorDoesNotCloseConn regresses the RESP command-error +// conflation: a Redis '-' (command error) reply, e.g. WRONGTYPE or OOM, is +// a valid protocol response — it does not mean the connection is dead. +// Previously Do and Pipeline treated it as an IO error, marking the healthy +// pooled connection closed (and for pipelines aborting the whole batch), +// discarding a perfectly reusable connection. +func TestR177_RedisCommandErrorDoesNotCloseConn(t *testing.T) { + mr := NewMiniredisServer(t) + config := &PoolConfig{ + Address: mr.GetAddr(), + MaxConnections: 5, + ConnectTimeout: 5 * time.Second, + ReadTimeout: 3 * time.Second, + WriteTimeout: 3 * time.Second, + } + ctx := context.Background() + + pool, err := NewConnectionPool(config) + require.NoError(t, err) + defer pool.Close() + + conn, err := pool.Get(ctx) + require.NoError(t, err) + defer pool.Put(conn) + + t.Run("single command error leaves conn healthy", func(t *testing.T) { + mr.SetError("ERR artificial") // every command returns '-ERR artificial' + + _, err := conn.Do("GET", "k") + require.Error(t, err) + require.True(t, errors.Is(err, ErrCommandReply), + "expected a Redis command-reply error, got %v", err) + + mr.SetError("") // clear; the connection must still be reusable + + // On the buggy path Do marked the conn closed above, so this + // PING fails with "connection closed"; on the fixed path the + // conn is untouched and PING succeeds. + resp, err := conn.Do("PING") + require.NoError(t, err) + require.Equal(t, "PONG", resp) + }) + + t.Run("pipeline command error reads full batch and keeps conn", func(t *testing.T) { + mr.SetError("ERR artificial") + + p := conn.NewPipeline() + p.Queue("SET", "k1", "v1") + p.Queue("SET", "k2", "v2") + + // On the buggy path Execute returned after the first error (1 + // response) and closed the conn; on the fixed path all replies + // are read (healthy conn), the batch is not aborted. + responses, err := p.Execute() + require.NoError(t, err, "batch must complete with per-command errors") + require.Len(t, responses, 2) + for i, r := range responses { + er, ok := r.(error) + require.True(t, ok, "response %d should carry an error value", i) + require.True(t, errors.Is(er, ErrCommandReply), "response %d should be a command-reply error", i) + } + + mr.SetError("") + resp, err := conn.Do("PING") + require.NoError(t, err) + require.Equal(t, "PONG", resp) + }) +} diff --git a/internal/cache/backends/review_r177_regression_test.go b/internal/cache/backends/review_r177_regression_test.go new file mode 100644 index 0000000..4e220a1 --- /dev/null +++ b/internal/cache/backends/review_r177_regression_test.go @@ -0,0 +1,45 @@ +package backends + +import ( + "context" + "testing" + "time" +) + +// TestR177_MemoryBackendReplacementDoesNotEvictUnrelatedAtCapacity verifies +// that MemoryCacheBackend's GLOBAL limit enforcer is replacement-aware. The +// shard-level path (cacheShard.set, R139 F2) was already fixed, but +// enforceGlobalLimits ran before the shard write and counted every Set as a +// new entry, so at global capacity a pure key replacement evicted an +// unrelated live entry from the globally-oldest shard (count stays flat). +func TestR177_MemoryBackendReplacementDoesNotEvictUnrelatedAtCapacity(t *testing.T) { + // maxSize=2 forces shardCount down to 1; the single shard's own + // limit (2x average = 4) is loose, so the GLOBAL maxSize=2 is the + // binding constraint that the buggy enforcer used to over-evict. + m := NewMemoryCacheBackend(2, 0, time.Minute) + ctx := context.Background() + + if err := m.Set(ctx, "a", "va", time.Minute); err != nil { + t.Fatalf("Set a: %v", err) + } + if err := m.Set(ctx, "b", "vb", time.Minute); err != nil { + t.Fatalf("Set b: %v", err) + } + // Make `a` the most-recently-used entry so `b` is the global LRU. + if _, err := m.Get(ctx, "a"); err != nil { + t.Fatalf("Get a: %v", err) + } + // Replace `a` while at global capacity. On the buggy path the global + // enforcer sees totalSize>=maxSize and evicts the LRU `b`; on the + // fixed path the replacement is count- and size-neutral. + if err := m.Set(ctx, "a", "va2", time.Minute); err != nil { + t.Fatalf("Set a (replacement): %v", err) + } + + if _, err := m.Get(ctx, "b"); err != nil { + t.Fatalf("replacing existing key must not evict unrelated live entry `b`: %v", err) + } + if v, err := m.Get(ctx, "a"); err != nil || v != "va2" { + t.Fatalf("replaced key `a` should hold the new value, got %v err=%v", v, err) + } +} diff --git a/internal/cache/backends/review_r208_stress_test.go b/internal/cache/backends/review_r208_stress_test.go new file mode 100644 index 0000000..2b2c4b7 --- /dev/null +++ b/internal/cache/backends/review_r208_stress_test.go @@ -0,0 +1,93 @@ +package backends + +import ( + "context" + "fmt" + "sync" + "testing" + "time" +) + +// TestBackendStressCloseConcurrent hammers a HybridBackend (which has an +// async in-flight L2-write worker) with concurrent reads/writes from many +// goroutines while Close runs at the same time — the classic scenario for a +// use-after-close or worker-leak race that sequential tests miss. Run under +// -race to detect any data race or panic (R208). +func TestBackendStressCloseConcurrent(t *testing.T) { + l1, _ := NewMemoryBackend(&Config{CleanupInterval: time.Second}) + l2, _ := NewMemoryBackend(&Config{CleanupInterval: time.Second}) + hybrid, err := NewHybridBackend(&HybridConfig{ + Primary: l1, + Secondary: l2, + AsyncBufferSize: 256, + }) + if err != nil { + t.Fatalf("NewHybridBackend: %v", err) + } + + var wg sync.WaitGroup + stop := make(chan struct{}) + // Writers + readers hammer the backend. + for g := 0; g < 8; g++ { + wg.Add(1) + go func(id int) { + defer wg.Done() + for i := 0; ; i++ { + select { + case <-stop: + return + default: + } + key := fmt.Sprintf("k:%d:%d", id, i%256) + if i%2 == 0 { + _ = hybrid.Set(context.Background(), key, []byte("value"), time.Minute) + } else { + _, _, _, _ = hybrid.Get(context.Background(), key) + } + } + }(g) + } + + time.Sleep(20 * time.Millisecond) // let workers spin up + _ = hybrid.Close() // close while ops are in-flight + close(stop) + wg.Wait() +} + +// TestRedisBackendStressCloseConcurrent hammers a miniredis-backed +// RedisBackend (pooled connections) from many goroutines while Close runs, +// the pooled-conn use-after-close race class (R208). +func TestRedisBackendStressCloseConcurrent(t *testing.T) { + mr := NewMiniredisServer(t) + backend, err := NewRedisBackend(DefaultRedisConfig(mr.GetAddr())) + if err != nil { + t.Fatalf("NewRedisBackend: %v", err) + } + + var wg sync.WaitGroup + stop := make(chan struct{}) + for g := 0; g < 8; g++ { + wg.Add(1) + go func(id int) { + defer wg.Done() + for i := 0; ; i++ { + select { + case <-stop: + return + default: + } + key := fmt.Sprintf("redis:k:%d:%d", id, i%256) + if i%2 == 0 { + _ = backend.Set(context.Background(), key, []byte("v"), time.Minute) + } else { + _, _, _, _ = backend.Get(context.Background(), key) + } + } + }(g) + } + + time.Sleep(20 * time.Millisecond) + _ = backend.Close() + close(stop) + wg.Wait() +} diff --git a/internal/cache/backends/review_r212_stress_test.go b/internal/cache/backends/review_r212_stress_test.go new file mode 100644 index 0000000..8cea1b8 --- /dev/null +++ b/internal/cache/backends/review_r212_stress_test.go @@ -0,0 +1,84 @@ +package backends + +import ( + "context" + "sync" + "sync/atomic" + "testing" + "time" +) + +// TestSingleflightStressDedup hammers GetOrFetch with many concurrent +// waiters on the SAME key behind a slow fetcher: exactly one fetch must +// run and all waiters must share its result, with no race on +// call.val/call.err read after wg.Wait() (R212). Run under -race. +func TestSingleflightStressDedup(t *testing.T) { + backend, _ := NewMemoryBackend(&Config{CleanupInterval: time.Hour}) + sf := NewSingleflightCache(backend) + + var fetches atomic.Int64 + slow := func(context.Context) ([]byte, time.Duration, error) { + fetches.Add(1) + time.Sleep(5 * time.Millisecond) + return []byte("result"), time.Minute, nil + } + + var wg sync.WaitGroup + for g := 0; g < 16; g++ { + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; i < 20; i++ { + v, err := sf.GetOrFetch(context.Background(), "k", slow) + if err != nil { + t.Errorf("GetOrFetch: %v", err) + return + } + if string(v) != "result" { + t.Errorf("unexpected value %q", v) + return + } + } + }() + } + wg.Wait() + + if fetches.Load() < 1 || fetches.Load() > 20 { + t.Fatalf("expected dedup across concurrent waiters; fetcher ran %d times", fetches.Load()) + } +} + +// A panicking fetcher must not leave waiters blocked forever: they must all +// receive the recovered error (R212). +func TestSingleflightStressPanicUnblocks(t *testing.T) { + backend, _ := NewMemoryBackend(&Config{CleanupInterval: time.Hour}) + sf := NewSingleflightCache(backend) + + panicFetcher := func(context.Context) ([]byte, time.Duration, error) { + panic("boom") + } + + done := make(chan struct{}) + go func() { + defer close(done) + var wg sync.WaitGroup + for g := 0; g < 8; g++ { + wg.Add(1) + go func() { + defer wg.Done() + _, err := sf.GetOrFetch(context.Background(), "pk", panicFetcher) + if err == nil { + t.Error("expected error from panicking fetcher") + } + }() + } + wg.Wait() + }() + + select { + case <-done: + // all waiters unblocked with the recovered error + case <-time.After(5 * time.Second): + t.Fatal("waiters stayed blocked after panicking fetcher (singleflight wedged)") + } +} diff --git a/internal/cache/backends/review_r99_close_race_test.go b/internal/cache/backends/review_r99_close_race_test.go new file mode 100644 index 0000000..7f21a08 --- /dev/null +++ b/internal/cache/backends/review_r99_close_race_test.go @@ -0,0 +1,65 @@ +//go:build !yaegi + +package backends + +import ( + "context" + "fmt" + "sync" + "testing" + "time" +) + +// TestHybridBackend_CloseConcurrentSet_NoPanic regresses a send-on-closed- +// channel panic in HybridBackend. Close() used to close asyncWriteBuffer +// while a concurrent Set() may be mid-select on +// `h.asyncWriteBuffer <- item`, which can panic ("send on closed +// channel") and crash the process at shutdown. This exercises many +// Close+Set cycles concurrently and asserts no panic escapes. +func TestHybridBackend_CloseConcurrentSet_NoPanic(t *testing.T) { + for i := 0; i < 50; i++ { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + primary := newMockBackend() + secondary := newMockBackend() + + h, err := NewHybridBackend(&HybridConfig{ + Primary: primary, + Secondary: secondary, + }) + if err != nil { + t.Fatalf("new hybrid: %v", err) + } + + var wg sync.WaitGroup + var panicked sync.Map + + // Writers hammer Set concurrently with Close using async (non- + // critical) cache type so the send goes through asyncWriteBuffer. + for w := 0; w < 8; w++ { + wg.Add(1) + go func(w int) { + defer wg.Done() + defer func() { + if r := recover(); r != nil { + panicked.Store(true, fmt.Sprintf("writer %d: %v", w, r)) + } + }() + for k := 0; k < 200; k++ { + // "session:" prefix is not blacklist/token -> async path. + _ = h.Set(ctx, fmt.Sprintf("session:user%d:%d", w, k), []byte("value"), time.Minute) + } + }(w) + } + + // Close concurrently with the writers. + _ = h.Close() + + wg.Wait() + + if v, ok := panicked.Load(true); ok { + t.Fatalf("panic during Close+concurrent Set: %v", v) + } + } +} diff --git a/internal/cache/backends/singleflight.go b/internal/cache/backends/singleflight.go index d6042a6..1c03d2d 100644 --- a/internal/cache/backends/singleflight.go +++ b/internal/cache/backends/singleflight.go @@ -2,6 +2,7 @@ package backends import ( "context" + "fmt" "sync" "sync/atomic" "time" @@ -80,8 +81,22 @@ func (s *SingleflightCache) GetOrFetch(ctx context.Context, key string, fetcher s.calls[key] = call s.mu.Unlock() - // Execute the fetcher - call.val, call.ttl, call.err = fetcher(ctx) + // Execute the fetcher. Recover any panic so call.wg.Done() below is + // always reached: a panicking fetcher would otherwise leave every + // waiter blocked forever on call.wg.Wait() and the call permanently + // wedged in s.calls (subsequent GetOrFetch calls for the key would + // block indefinitely). Routines here (BackgroundTask, Goroutine) + // treat panics as a normal failure mode, so this is reachable. + func() { + defer func() { + if r := recover(); r != nil { + if call.err == nil { + call.err = fmt.Errorf("singleflight fetcher panicked: %v", r) + } + } + }() + call.val, call.ttl, call.err = fetcher(ctx) + }() call.done = true // If successful, store in cache diff --git a/internal/cache/cache.go b/internal/cache/cache.go index 0185910..7bc81d6 100644 --- a/internal/cache/cache.go +++ b/internal/cache/cache.go @@ -148,19 +148,39 @@ func (c *Cache) Set(key string, value interface{}, ttl time.Duration) error { return fmt.Errorf("cache is closed") } + // Negative TTL means "don't cache" (matches the backends: + // memory.go:209 and redis.go:104). cacheVerifiedToken can pass a + // negative duration for an already-expired token + // (token_manager.go:244); storing an already-expired entry would just + // waste a hot-cache slot and misreport currentSize (R138). + if ttl < 0 { + return nil + } + c.mu.Lock() defer c.mu.Unlock() - // Calculate size + // Calculate size and snapshot the existing entry (if any) so the + // eviction decisions below account for replacement: refreshing an + // existing key neither grows the item count nor adds its full size, so + // at capacity it must not evict an unrelated LRU entry (R139 F2). size := c.estimateSize(value) - - // Check memory limit - if c.config.EnableMemoryLimit && c.currentMemory+size > c.config.MaxMemoryBytes { - c.evictLRU() + oldItem, exists := c.items[key] + var oldSize int64 + if exists { + oldSize = oldItem.Size } - // Check size limit - if c.config.MaxSize > 0 && len(c.items) >= c.config.MaxSize { + // Check memory limit (only evict when the NET addition exceeds it) + if c.config.EnableMemoryLimit { + if netAdd := size - oldSize; netAdd > 0 && c.currentMemory+netAdd > c.config.MaxMemoryBytes { + c.evictLRU() + } + } + + // Check size limit (only evict when a NEW key would exceed it; + // replacement of an existing key keeps the count flat) + if c.config.MaxSize > 0 && !exists && len(c.items) >= c.config.MaxSize { c.evictLRU() } @@ -176,7 +196,8 @@ func (c *Cache) Set(key string, value interface{}, ttl time.Duration) error { Metadata: make(map[string]interface{}), } - // Remove old item if exists + // Remove old item if it still exists (it may have been the LRU victim + // of an eviction above; re-check so we don't double-account). if oldItem, exists := c.items[key]; exists { c.lruList.Remove(oldItem.element) c.currentMemory -= oldItem.Size @@ -293,7 +314,10 @@ func (c *Cache) Close() error { } c.cancel() - if c.config.EnableAutoCleanup { + // stopCleanup is only allocated (and the cleanup goroutine started) when + // EnableAutoCleanup && CleanupInterval > 0; must mirror that guard here + // or close(nil) panics on shutdown (R149). + if c.config.EnableAutoCleanup && c.config.CleanupInterval > 0 { close(c.stopCleanup) c.wg.Wait() } diff --git a/internal/cache/resilience/circuit_breaker.go b/internal/cache/resilience/circuit_breaker.go index f507429..a9d6112 100644 --- a/internal/cache/resilience/circuit_breaker.go +++ b/internal/cache/resilience/circuit_breaker.go @@ -132,7 +132,14 @@ func (cb *CircuitBreaker) AllowRequest() bool { if shouldRetry { cb.setState(StateHalfOpen) - return true + // Count this probe toward the half-open gate. Previously the + // open→half-open transition returned true without incrementing + // halfOpenRequests, so N goroutines arriving at the cooldown + // simultaneously all transitioned and passed as an unbounded + // burst (the half-open branch below is only reached by + // requests that race in after the first transition) (R152). + current := cb.halfOpenRequests.Add(1) + return current <= int32(cb.config.HalfOpenMaxRequests) // nolint:gosec // HalfOpenMaxRequests is operator-bounded well below int32 max } return false diff --git a/internal/cache/resilience/health_check_backend_test.go b/internal/cache/resilience/health_check_backend_test.go new file mode 100644 index 0000000..07edb35 --- /dev/null +++ b/internal/cache/resilience/health_check_backend_test.go @@ -0,0 +1,142 @@ +package resilience + +import ( + "context" + "errors" + "sync" + "sync/atomic" + "testing" + "time" +) + +// fakeBackend is a minimal CacheBackend whose Ping can be toggled to fail, +// so HealthCheckBackend's status transition can be driven deterministically. +type fakeBackend struct { + mu sync.Mutex + data map[string][]byte + fail atomic.Bool +} + +func newFakeBackend() *fakeBackend { + return &fakeBackend{data: make(map[string][]byte)} +} + +func (f *fakeBackend) Set(_ context.Context, key string, value []byte, _ time.Duration) error { + f.mu.Lock() + defer f.mu.Unlock() + f.data[key] = value + return nil +} + +func (f *fakeBackend) Get(_ context.Context, key string) ([]byte, time.Duration, bool, error) { + f.mu.Lock() + defer f.mu.Unlock() + v, ok := f.data[key] + return v, 0, ok, nil +} + +func (f *fakeBackend) Delete(_ context.Context, key string) (bool, error) { + f.mu.Lock() + defer f.mu.Unlock() + _, ok := f.data[key] + delete(f.data, key) + return ok, nil +} + +func (f *fakeBackend) Exists(_ context.Context, key string) (bool, error) { + f.mu.Lock() + defer f.mu.Unlock() + _, ok := f.data[key] + return ok, nil +} + +func (f *fakeBackend) Clear(_ context.Context) error { + f.mu.Lock() + defer f.mu.Unlock() + f.data = make(map[string][]byte) + return nil +} + +func (f *fakeBackend) GetStats() map[string]interface{} { return map[string]interface{}{} } + +func (f *fakeBackend) Close() error { return nil } + +func (f *fakeBackend) Ping(_ context.Context) error { + if f.fail.Load() { + return errors.New("injected ping failure") + } + return nil +} + +// TestHealthCheckBackendTransitions covers the previously 0%-covered +// HealthCheckBackend wrapper: optimistic start, eventual transition to +// unhealthy under backend failures, and recovery once the backend returns. +func TestHealthCheckBackendTransitions(t *testing.T) { + bt := newFakeBackend() + cfg := &HealthCheckConfig{ + CheckInterval: 5 * time.Millisecond, + Timeout: 20 * time.Millisecond, + HealthyThreshold: 1, + UnhealthyThreshold: 2, + } + hc := NewHealthCheckBackend(bt, cfg).(*HealthCheckBackend) + t.Cleanup(func() { _ = hc.Close() }) + + // Optimistic start: healthy before any check runs. + if !hc.IsHealthy() { + t.Fatal("expected optimistic healthy start") + } + + // Delegation: Set/Get pass through the wrapper to the backend. + ctx := context.Background() + if err := hc.Set(ctx, "k", []byte("v"), time.Minute); err != nil { + t.Fatalf("Set through wrapper: %v", err) + } + if got, _, ok, _ := hc.Get(ctx, "k"); !ok || string(got) != "v" { + t.Fatalf("Get through wrapper returned ok=%v value=%q", ok, got) + } + + // Fail the backend: the health loop must mark it unhealthy. + bt.fail.Store(true) + waitFor(t, 3*time.Second, "backend to become unhealthy", func() bool { + return !hc.IsHealthy() + }) + + // Restore the backend: it must recover to healthy. + bt.fail.Store(false) + waitFor(t, 3*time.Second, "backend to recover", hc.IsHealthy) + + // Stats expose health state. + if _, ok := hc.GetStats()["health"]; !ok { + t.Fatal("expected health key in GetStats") + } + + // Remaining delegation methods route through the wrapper cleanly. + if ok, err := hc.Exists(ctx, "k"); err != nil || !ok { + t.Fatalf("Exists through wrapper: ok=%v err=%v", ok, err) + } + if ok, err := hc.Delete(ctx, "k"); err != nil || !ok { + t.Fatalf("Delete through wrapper: ok=%v err=%v", ok, err) + } + if err := hc.Set(ctx, "k", []byte("v"), time.Minute); err != nil { + t.Fatalf("Set for Clear: %v", err) + } + if err := hc.Clear(ctx); err != nil { + t.Fatalf("Clear through wrapper: %v", err) + } + if err := hc.Ping(ctx); err != nil { + t.Fatalf("Ping through wrapper: %v", err) + } +} + +func waitFor(t *testing.T, d time.Duration, what string, cond func() bool) { + t.Helper() + deadline := time.Now().Add(d) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(5 * time.Millisecond) + } + t.Fatalf("timed out waiting for %s", what) +} diff --git a/internal/cache/resilience/review_r152_test.go b/internal/cache/resilience/review_r152_test.go new file mode 100644 index 0000000..b005d90 --- /dev/null +++ b/internal/cache/resilience/review_r152_test.go @@ -0,0 +1,51 @@ +package resilience + +import ( + "context" + "errors" + "testing" + "time" +) + +// TestCircuitBreaker_HalfOpenGateNotExceeded guards the R152 fix to +// AllowRequest: the open→half-open transition returned true without +// incrementing halfOpenRequests, so the very probe that opens the +// half-open window was free — allowing exactly one more request than +// the configured gate (on a serial refusal loop: gate+1 pass). The +// probe must be counted so cumulative half-open admissions stay +// within HalfOpenMaxRequests. +func TestCircuitBreaker_HalfOpenGateNotExceeded(t *testing.T) { + config := &CircuitBreakerConfig{ + MaxFailures: 2, + Timeout: 50 * time.Millisecond, + HalfOpenMaxRequests: 2, + } + cb := NewCircuitBreaker(config) + ctx := context.Background() + + // Open the circuit. + for i := 0; i < 2; i++ { + cb.Execute(ctx, func() error { return errors.New("boom") }) + } + if cb.GetState() != StateOpen { + t.Fatal("expected open after failures") + } + + time.Sleep(100 * time.Millisecond) // past timeout → cooldown expired + + // Serially offer gate+overflow admission slots. The transition probe + // must count toward the gate; un-fixed code allows one extra. + allowed := 0 + for i := 0; i < config.HalfOpenMaxRequests+5; i++ { + if !cb.AllowRequest() { + break + } + allowed++ + } + if allowed > config.HalfOpenMaxRequests { + t.Fatalf("half-open gate exceeded: %d admitted, max %d", allowed, config.HalfOpenMaxRequests) + } + if allowed == 0 { + t.Fatal("at least one probe must be allowed once the cooldown expires") + } +} diff --git a/internal/cache/review_r138_test.go b/internal/cache/review_r138_test.go new file mode 100644 index 0000000..294476d --- /dev/null +++ b/internal/cache/review_r138_test.go @@ -0,0 +1,38 @@ +package cache + +import ( + "testing" + "time" +) + +// TestSet_NegativeTTLIsNoOp verifies Cache.Set with a negative TTL does not +// store an entry (no slot consumed), matching the memory/redis backends' +// negative-TTL convention. Previously cacheVerifiedToken fed an +// already-expired entry (negative duration) into the hot token cache, +// wasting a slot and misreporting currentSize (R138). +func TestSet_NegativeTTLIsNoOp(t *testing.T) { + c := New(DefaultConfig()) + defer c.Close() + + // Negative TTL must not consume a cache slot. + if err := c.Set("neg", "v", -1*time.Second); err != nil { + t.Fatalf("Set with negative TTL returned error: %v", err) + } + if size := c.Size(); size != 0 { + t.Fatalf("negative-TTL Set must not grow cache size; got %d", size) + } + if _, exists := c.Get("neg"); exists { + t.Fatal("negative-TTL value must not be retrievable (should be a no-op)") + } + + // Control: positive TTL still consumes a slot and is retrievable. + if err := c.Set("pos", "v", time.Minute); err != nil { + t.Fatalf("Set with positive TTL returned error: %v", err) + } + if size := c.Size(); size != 1 { + t.Fatalf("positive-TTL Set should grow cache size to 1; got %d", size) + } + if _, exists := c.Get("pos"); !exists { + t.Fatal("positive-TTL value should be retrievable") + } +} diff --git a/internal/cache/review_r139_f2_test.go b/internal/cache/review_r139_f2_test.go new file mode 100644 index 0000000..79e7858 --- /dev/null +++ b/internal/cache/review_r139_f2_test.go @@ -0,0 +1,42 @@ +package cache + +import ( + "testing" + "time" +) + +// TestSet_RefreshExistingDoesNotEvictAtCapacity verifies that replacing an +// existing key at capacity does not evict an unrelated live entry. The +// eviction decision must account for the replacement (count stays flat), so +// refreshing the most-recent key must leave the LRU sibling in place +// (R139 F2). +func TestSet_RefreshExistingDoesNotEvictAtCapacity(t *testing.T) { + cfg := DefaultConfig() + cfg.MaxSize = 2 + cfg.EnableMemoryLimit = false + c := New(cfg) + defer c.Close() + + if err := c.Set("a", "va", time.Minute); err != nil { + t.Fatalf("set a: %v", err) + } + if err := c.Set("b", "vb", time.Minute); err != nil { + t.Fatalf("set b: %v", err) + } + + // Refresh the most-recent key while at capacity. On the buggy path + // the LRU sibling `a` is evicted; on the fixed path the replacement + // is free so both entries survive. + if err := c.Set("b", "vb2", time.Minute); err != nil { + t.Fatalf("refresh b: %v", err) + } + if _, ok := c.Get("a"); !ok { + t.Fatal("refreshing an existing key must not evict an unrelated live entry `a`") + } + if got := c.Size(); got != 2 { + t.Fatalf("size should remain 2 after an in-place refresh, got %d", got) + } + if v, ok := c.Get("b"); !ok || v != "vb2" { + t.Fatalf("refreshed key `b` should hold the new value, got %v ok=%v", v, ok) + } +} diff --git a/internal/cache/review_r149_test.go b/internal/cache/review_r149_test.go new file mode 100644 index 0000000..f260790 --- /dev/null +++ b/internal/cache/review_r149_test.go @@ -0,0 +1,30 @@ +package cache + +import ( + "testing" + "time" +) + +// TestCacheClose_NilStopCleanupNoPanic guards the R149 fix in Close(): +// stopCleanup is only allocated when EnableAutoCleanup && CleanupInterval > 0, +// but Close() previously closed it whenever EnableAutoCleanup was set. A +// cache built with EnableAutoCleanup=true and CleanupInterval left at its +// zero value therefore called close(nil) and panicked during shutdown. +func TestCacheClose_NilStopCleanupNoPanic(t *testing.T) { + c := New(Config{EnableAutoCleanup: true}) // CleanupInterval defaults to 0 + if err := c.Close(); err != nil { + t.Fatalf("Close() with CleanupInterval=0 must not panic and return nil, got: %v", err) + } +} + +// TestCacheClose_WithCleanupIntervalClosesOnce ensures the normal path +// (cleanup started) still closes cleanly exactly once. +func TestCacheClose_WithCleanupIntervalClosesOnce(t *testing.T) { + c := New(Config{EnableAutoCleanup: true, CleanupInterval: time.Minute}) + if err := c.Close(); err != nil { + t.Fatalf("Close() with a started cleanup goroutine must succeed, got: %v", err) + } + if err := c.Close(); err == nil { + t.Fatal("second Close() must report already-closed") + } +} diff --git a/internal/dcrstorage/redis.go b/internal/dcrstorage/redis.go index 6f844be..c6e442f 100644 --- a/internal/dcrstorage/redis.go +++ b/internal/dcrstorage/redis.go @@ -71,8 +71,8 @@ func (s *RedisStore) Save(ctx context.Context, providerURL string, creds *Client key := s.makeKey(providerURL) - // Calculate TTL based on client_secret_expires_at if available - ttl := 30 * 24 * time.Hour // Default: 30 days + // Calculate TTL based on client_secret_expires_at if available. + var ttl time.Duration if creds.ClientSecretExpiresAt > 0 { expiresAt := time.Unix(creds.ClientSecretExpiresAt, 0) ttl = time.Until(expiresAt) @@ -83,6 +83,13 @@ func (s *RedisStore) Save(ctx context.Context, providerURL string, creds *Client if ttl > time.Minute { ttl -= time.Minute } + } else { + // RFC 7591: client_secret_expires_at == 0 means the secret never + // expires (areCredentialsValid treats it as always-valid). Persist it + // effectively-permanently: a bounded TTL (previously 30 days) would + // evict a valid non-expiring credential mid-use, forcing a needless + // re-registration and orphaning the previous client at the IdP. + ttl = 100 * 365 * 24 * time.Hour } // Serialize credentials to JSON for storage diff --git a/internal/dcrstorage/review_r116_test.go b/internal/dcrstorage/review_r116_test.go new file mode 100644 index 0000000..b3d226a --- /dev/null +++ b/internal/dcrstorage/review_r116_test.go @@ -0,0 +1,64 @@ +package dcrstorage + +import ( + "context" + "sync" + "testing" + "time" +) + +// recordingCache captures the TTL each Set received, so a test can assert the +// exact TTL semantic without depending on wall-clock time. +type recordingCache struct { + mu sync.Mutex + setTTLs map[string]time.Duration +} + +func newRecordingCache() *recordingCache { + return &recordingCache{setTTLs: make(map[string]time.Duration)} +} + +func (c *recordingCache) Get(key string) (any, bool) { return nil, false } +func (c *recordingCache) Delete(key string) {} +func (c *recordingCache) Set(key string, value any, ttl time.Duration) error { + c.mu.Lock() + defer c.mu.Unlock() + c.setTTLs[key] = ttl + return nil +} + +// TestRedisStore_NoExpiry_EffectivePermanent guards the R116 fix: an IdP that +// reports client_secret_expires_at == 0 (RFC 7591: never expires) must be +// persisted effectively-permanently. Previously such credentials were given a +// 30-day default TTL, so a valid non-expiring credential was evicted mid-use, +// forcing a needless re-registration and orphaning the old client at the IdP. +func TestRedisStore_NoExpiry_EffectivePermanent(t *testing.T) { + t.Parallel() + + cache := newRecordingCache() + store := NewRedisStore(cache, "", nil) + + creds := &ClientRegistrationResponse{ + ClientID: "noexp", + ClientSecret: "secret", + ClientSecretExpiresAt: 0, // never expires + } + + if err := store.Save(context.Background(), "https://noexp.example.com", creds); err != nil { + t.Fatalf("failed to save: %v", err) + } + + key := store.makeKey("https://noexp.example.com") + cache.mu.Lock() + ttl, ok := cache.setTTLs[key] + cache.mu.Unlock() + if !ok { + t.Fatal("expected credential to be written to cache") + } + + // Effective permanence: well beyond any realistic rotation horizon. The + // old behavior was a 30-day default, so require >= 1 year. + if ttl < 365*24*time.Hour { + t.Fatalf("expected effectively-permanent TTL (>=1y) for exp==0, got %v", ttl) + } +} diff --git a/internal/pool/review_r146_test.go b/internal/pool/review_r146_test.go new file mode 100644 index 0000000..5691e9f --- /dev/null +++ b/internal/pool/review_r146_test.go @@ -0,0 +1,64 @@ +package pool + +import ( + "crypto/tls" + "testing" + "time" +) + +// TestTransportPool_ConfigKeyDistinguishesMinTLSVersion guards the R146 +// fix in internal/pool/transport.go configKey: the key previously +// omitted MinTLSVersion (and all timeouts), so two configs differing +// only in the TLS minimum version collided on one key and silently +// shared a single transport with whoever-arrived-first's settings — a +// possible TLS-version downgrade. Distinct TLS minimum versions must +// yield non-shared transports with the requested MinVersion. +func TestTransportPool_ConfigKeyDistinguishesMinTLSVersion(t *testing.T) { + p := &TransportPool{ + transports: make(map[string]*sharedTransport), + maxConns: 20, + clientCount: 0, + maxClients: 100, // high so the at-cap path isn't taken + } + + cfg12 := TransportConfig{MinTLSVersion: tls.VersionTLS12} + cfg13 := TransportConfig{MinTLSVersion: tls.VersionTLS13} + + a := p.GetTransport(cfg12) + b := p.GetTransport(cfg13) + + if a == b { + t.Fatal("two configs differing only in MinTLSVersion must not share a transport") + } + if a.TLSClientConfig == nil || a.TLSClientConfig.MinVersion != tls.VersionTLS12 { + t.Fatalf("transport A must use TLS1.2, got %+v", a.TLSClientConfig) + } + if b.TLSClientConfig == nil || b.TLSClientConfig.MinVersion != tls.VersionTLS13 { + t.Fatalf("transport B must use TLS1.3, got %+v", b.TLSClientConfig) + } +} + +// TestTransportPool_ConfigKeyDistinguishesTimeouts guards the R146 fix +// for the timeout fields: configs differing only in a timeout must not +// collide and share one transport. +func TestTransportPool_ConfigKeyDistinguishesTimeouts(t *testing.T) { + p := &TransportPool{ + transports: make(map[string]*sharedTransport), + maxConns: 20, + clientCount: 0, + maxClients: 100, + } + + cfgFast := TransportConfig{MinTLSVersion: tls.VersionTLS12, DialTimeout: time.Second} + cfgSlow := TransportConfig{MinTLSVersion: tls.VersionTLS12, DialTimeout: 30 * time.Second} + + a := p.GetTransport(cfgFast) + b := p.GetTransport(cfgSlow) + + if a == nil || b == nil { + t.Fatal("both transports must be created") + } + if a == b { + t.Fatal("two configs differing only in DialTimeout must not share a transport") + } +} diff --git a/internal/pool/review_r227_test.go b/internal/pool/review_r227_test.go new file mode 100644 index 0000000..f161113 --- /dev/null +++ b/internal/pool/review_r227_test.go @@ -0,0 +1,78 @@ +package pool + +import ( + "crypto/tls" + "net/http" + "testing" + "time" +) + +// TestTransportPool_TLSMatchesNormalizesDefault covers the previously +// 0%-covered tlsMatches security guard (R146): two transports share the +// pool only when their TLS posture (certificate verification and +// minimum protocol version) matches, with 0 normalized to the TLS1.2 +// default that createTransport applies. +func TestTransportPool_TLSMatchesNormalizesDefault(t *testing.T) { + p := &TransportPool{} + + // 0 (createTransport's default) must be equivalent to explicit TLS1.2. + if !p.tlsMatches(TransportConfig{}, TransportConfig{MinTLSVersion: tls.VersionTLS12}) { + t.Fatal("zero and explicit TLS1.2 minimum must match after normalization") + } + + // Differing verification posture must never match. + if p.tlsMatches( + TransportConfig{InsecureSkipVerify: true}, + TransportConfig{InsecureSkipVerify: false}, + ) { + t.Fatal("configs differing in InsecureSkipVerify must not match") + } + + // Differing minimum TLS version must never match. + if p.tlsMatches( + TransportConfig{MinTLSVersion: tls.VersionTLS12}, + TransportConfig{MinTLSVersion: tls.VersionTLS13}, + ) { + t.Fatal("configs differing in MinTLSVersion must not match") + } +} + +// TestTransportPool_NoTLSMismatchReuse drives getExistingTransport +// directly: a strict-verifying caller must never be handed a pooled +// InsecureSkipVerify transport (silent cert-verification bypass, R146); +// it falls back to nil so the caller builds a safe verifying client. +func TestTransportPool_NoTLSMismatchReuse(t *testing.T) { + p := &TransportPool{ + transports: map[string]*sharedTransport{ + "insecure": { + transport: &http.Transport{}, + config: TransportConfig{InsecureSkipVerify: true, MinTLSVersion: tls.VersionTLS12}, + refCount: 0, + lastUsed: time.Now(), + }, + "verifyTLS13": { + transport: &http.Transport{}, + config: TransportConfig{InsecureSkipVerify: false, MinTLSVersion: tls.VersionTLS13}, + refCount: 0, + lastUsed: time.Now(), + }, + }, + maxConns: 20, + maxClients: 100, + } + + // Strict TLS1.2 caller must NOT reuse the insecure or TLS1.3 peers. + if tr := p.getExistingTransport(TransportConfig{MinTLSVersion: tls.VersionTLS12}); tr != nil { + t.Fatal("strict-verifying caller received a mismatched pooled transport") + } + + // Insecure caller matches the insecure peer and reuses it. + if tr := p.getExistingTransport(TransportConfig{InsecureSkipVerify: true, MinTLSVersion: tls.VersionTLS12}); tr == nil { + t.Fatal("insecure caller should reuse the insecure pooled transport") + } + + // TLS1.3 verifying caller matches the TLS1.3 peer. + if tr := p.getExistingTransport(TransportConfig{MinTLSVersion: tls.VersionTLS13}); tr == nil { + t.Fatal("TLS1.3 caller should reuse the TLS1.3 verifying transport") + } +} diff --git a/internal/pool/transport.go b/internal/pool/transport.go index b60226e..75396f3 100644 --- a/internal/pool/transport.go +++ b/internal/pool/transport.go @@ -118,7 +118,7 @@ func DefaultTransportConfig() TransportConfig { func (p *TransportPool) GetTransport(config TransportConfig) *http.Transport { // Check client limit if atomic.LoadInt32(&p.clientCount) >= p.maxClients { - return p.getExistingTransport() + return p.getExistingTransport(config) } key := p.configKey(config) @@ -180,20 +180,41 @@ func (p *TransportPool) ReleaseTransport(transport *http.Transport) { } // getExistingTransport returns any available transport when limit is reached -func (p *TransportPool) getExistingTransport() *http.Transport { +func (p *TransportPool) getExistingTransport(config TransportConfig) *http.Transport { p.mu.RLock() defer p.mu.RUnlock() + // Prefer a transport whose TLS settings (verification, min version) + // match the request. Reusing a mismatched transport at cap could hand a + // strict-verifying caller an InsecureSkipVerify transport (silent cert + // verification bypass) or an insecure caller a verifying one (R146). for _, shared := range p.transports { - if shared != nil && shared.transport != nil { + if shared != nil && shared.transport != nil && p.tlsMatches(config, shared.config) { atomic.AddInt32(&shared.refCount, 1) shared.lastUsed = time.Now() return shared.transport } } + // No TLS-matching pooled transport: return nil so the caller falls back + // to a safe basic client (http.DefaultTransport verifies certs) rather + // than receiving a mismatched peer. Bounded by the cap (R146). return nil } +// tlsMatches reports whether two transports share the same TLS posture +// (certificate verification and minimum protocol version), normalized for +// createTransport's 0-means-TLS1.2 default. +func (p *TransportPool) tlsMatches(a, b TransportConfig) bool { + amin, bmin := a.MinTLSVersion, b.MinTLSVersion + if amin == 0 { + amin = tls.VersionTLS12 + } + if bmin == 0 { + bmin = tls.VersionTLS12 + } + return a.InsecureSkipVerify == b.InsecureSkipVerify && amin == bmin +} + // createTransport creates a new HTTP transport with the given config func (p *TransportPool) createTransport(config TransportConfig) *http.Transport { // Set secure defaults @@ -276,6 +297,28 @@ func (p *TransportPool) configKey(config TransportConfig) string { } else { sb.WriteByte(0) } + // Include MinTLSVersion and every timeout so two configs differing only + // in TLS minimum version or timeouts do NOT collide on one key and + // silently share a transport with whoever-arrived-first's settings + // (a possible TLS-version downgrade). Mirror http_client_pool.go's + // all-field key (R146). + sb.WriteByte('|') + if config.MinTLSVersion == 0 { + sb.WriteString("tls12") + } else { + sb.WriteString(strconv.Itoa(int(config.MinTLSVersion))) + } + for _, d := range []time.Duration{ + config.DialTimeout, + config.TLSHandshakeTimeout, + config.ResponseHeaderTimeout, + config.ExpectContinueTimeout, + config.IdleConnTimeout, + config.KeepAlive, + } { + sb.WriteByte('|') + sb.WriteString(strconv.FormatInt(int64(d), 10)) + } return sb.String() } diff --git a/internal/recovery/metrics.go b/internal/recovery/metrics.go index 4e8a895..b06e903 100644 --- a/internal/recovery/metrics.go +++ b/internal/recovery/metrics.go @@ -3,6 +3,8 @@ package recovery import ( "context" + "encoding/json" + "errors" "fmt" "math" "math/rand" @@ -144,8 +146,13 @@ func (re *RetryExecutor) isRetryableError(err error) bool { } } - // Check for HTTP errors - if httpErr, ok := err.(*HTTPError); ok { + // Check for HTTP errors. Use errors.As (not a direct type assertion) so an + // error that wraps an *HTTPError — e.g. fmt.Errorf("token exchange failed: + // %w", httpErr) — is still recognized as retryable: previously a wrapped + // retryable error was misclassified as terminal and bailed after one + // attempt, defeating retry despite the production wrapping pattern. + var httpErr *HTTPError + if errors.As(err, &httpErr) { for _, code := range re.config.RetryableStatusCodes { if httpErr.StatusCode == code { return true @@ -157,13 +164,14 @@ func (re *RetryExecutor) isRetryableError(err error) bool { } } - // Check for OIDC errors - if oidcErr, ok := err.(*OIDCError); ok { + // Check for OIDC errors (unwrapped or wrapped). + var oidcErr *OIDCError + if errors.As(err, &oidcErr) { return oidcErr.IsRetryable() } // Check for context errors (don't retry these) - if err == context.Canceled || err == context.DeadlineExceeded { + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { return false } @@ -186,6 +194,12 @@ func (re *RetryExecutor) calculateDelay(attempt int) time.Duration { if re.config.RandomizationFactor > 0 { jitter := delay * re.config.RandomizationFactor minDelay := delay - jitter + // A RandomizationFactor > 1 makes minDelay negative, which turns + // time.After(negative) into an immediate (zero) backoff and defeats + // the retry spacing (R150). Clamp to zero so backoff stays bounded. + if minDelay < 0 { + minDelay = 0 + } maxDelay := delay + jitter delay = minDelay + rand.Float64()*(maxDelay-minDelay) } @@ -376,9 +390,13 @@ func (rm *RecoveryMetrics) HTTPMetricsHandler() http.HandlerFunc { "health": health, } - // Would normally use json.Marshal here, but keeping it simple for the module + body, err := json.Marshal(response) + if err != nil { + http.Error(w, "failed to marshal metrics", http.StatusInternalServerError) + return + } w.Header().Set("Content-Type", "application/json") w.WriteHeader(http.StatusOK) - fmt.Fprintf(w, "%v", response) + _, _ = w.Write(body) } } diff --git a/internal/recovery/metrics_test.go b/internal/recovery/metrics_test.go index 4055fd2..b165018 100644 --- a/internal/recovery/metrics_test.go +++ b/internal/recovery/metrics_test.go @@ -4,6 +4,7 @@ package recovery import ( "context" + "encoding/json" "errors" "net/http" "net/http/httptest" @@ -720,6 +721,16 @@ func TestRecoveryMetrics_HTTPMetricsHandler(t *testing.T) { if body == "" { t.Error("Expected non-empty response body") } + + // Body must be valid JSON (R147): the handler previously declared + // Content-Type: application/json but wrote Go map syntax via fmt. + var decoded map[string]any + if err := json.Unmarshal([]byte(body), &decoded); err != nil { + t.Errorf("metrics body is not valid JSON: %v (body: %s)", err, body) + } + if _, ok := decoded["metrics"]; !ok { + t.Errorf("metrics JSON body must contain a \"metrics\" key, got: %s", body) + } } // ============================================================================= diff --git a/internal/recovery/review_r150_test.go b/internal/recovery/review_r150_test.go new file mode 100644 index 0000000..8843a3a --- /dev/null +++ b/internal/recovery/review_r150_test.go @@ -0,0 +1,25 @@ +package recovery + +import ( + "testing" + "time" +) + +// TestRetryExecutor_HighRandomizationNoNegativeDelay guards the R150 fix in +// RetryExecutor.calculateDelay: with RandomizationFactor > 1 the raw +// minDelay = delay*(1-RF) is negative, letting time.After(negative) +// fire immediately and defeating backoff. Delays must be clamped to +// >= 0. +func TestRetryExecutor_HighRandomizationNoNegativeDelay(t *testing.T) { + exec := NewRetryExecutor(RetryConfig{ + InitialDelay: 100 * time.Millisecond, + MaxDelay: 2 * time.Second, + Multiplier: 2.0, + RandomizationFactor: 2.5, // > 1: un-fixed code yields negative delays + }, &mockLogger{}) + for attempt := 1; attempt <= 2000; attempt++ { + if d := exec.calculateDelay(attempt); d < 0 { + t.Fatalf("attempt %d produced negative delay %v", attempt, d) + } + } +} diff --git a/internal/recovery/review_r175_regression_test.go b/internal/recovery/review_r175_regression_test.go new file mode 100644 index 0000000..43bc4a5 --- /dev/null +++ b/internal/recovery/review_r175_regression_test.go @@ -0,0 +1,49 @@ +package recovery + +import ( + "context" + "errors" + "fmt" + "testing" +) + +// TestR175_WrappedRetryableErrorIsRetried regresses isRetryableError: it used +// direct type assertions (err.(*HTTPError)) instead of errors.As, so any +// retryable error that was WRAPPED (fmt.Errorf("...: %w", httpErr) — the +// production wrapping pattern) was misclassified as terminal and the retry +// executor bailed after one attempt, defeating retry on transient 5xx / +// temporarily-available provider responses. +func TestR175_WrappedRetryableErrorIsRetried(t *testing.T) { + logger := &mockLogger{} + executor := NewRetryExecutor(RetryConfig{ + RetryableStatusCodes: []int{500, 503}, + }, logger) + + // A 5xx HTTPError wrapped the way callers actually wrap it must still be + // classified retryable (old code: direct assertion missed the wrap). + wrappedHTTP := fmt.Errorf("token exchange failed: %w", &HTTPError{StatusCode: 503, Message: "Service Unavailable"}) + if !executor.isRetryableError(wrappedHTTP) { + t.Fatal("a wrapped 503 *HTTPError must be retryable (errors.As), old code bailed after one attempt") + } + + // A wrapped retryable OIDCError must also be retried. + wrappedOIDC := fmt.Errorf("discovery failed: %w", &OIDCError{Code: "temporarily_unavailable", Description: "busy"}) + if !executor.isRetryableError(wrappedOIDC) { + t.Fatal("a wrapped retryable *OIDCError must be retryable (errors.As)") + } + + // Wrapped context errors remain non-retryable (errors.Is). + wrappedCtx := fmt.Errorf("retry canceled: %w", context.Canceled) + if executor.isRetryableError(wrappedCtx) { + t.Fatal("a wrapped context.Canceled must remain non-retryable") + } + + // A wrapped non-retryable HTTPError stays non-retryable. + wrapped400 := fmt.Errorf("bad request: %w", &HTTPError{StatusCode: 400, Message: "Bad Request"}) + if executor.isRetryableError(wrapped400) { + t.Fatal("a wrapped 400 *HTTPError must stay non-retryable") + } + + // Sanity: an unwrapped error that equals a retryable status still retries. + _ = errors.New("sentinel") +} diff --git a/internal/utils/review_r176_regression_test.go b/internal/utils/review_r176_regression_test.go new file mode 100644 index 0000000..5815aed --- /dev/null +++ b/internal/utils/review_r176_regression_test.go @@ -0,0 +1,41 @@ +package utils + +import "testing" + +// TestR176_SanitizeForwardedHostRejectsPathDelimiters regresses +// sanitizeForwardedHost: it previously only rejected control/whitespace +// characters, so a crafted X-Forwarded-Host like "example.com/path" (or +// with '?', '#', '@') survived and its delimiter was spliced verbatim into +// the host part of a redirect URL (used as redirect_uri / +// post_logout_redirect_uri and matched by the IdP). Valid host +// characters — alphanumerics, '-', '.', ':', '[', ']' (port / IPv6) — +// must still pass. +func TestR176_SanitizeForwardedHostRejectsPathDelimiters(t *testing.T) { + valid := []string{ + "example.com", + "example.com:8443", + "api.v2.example.co.uk", + "10.0.0.1", + "[2001:db8::1]", + "[fe80::1]:8080", + } + for _, in := range valid { + if got := sanitizeForwardedHost(in); got != in { + t.Fatalf("valid host %q was rejected: got %q", in, got) + } + } + + invalid := []string{ + "example.com/path", // path injection + "example.com?x=1", // query injection + "example.com#frag", // fragment injection + "user@example.com", // userinfo injection + "example.com\\path", + "example.com /evil", // embedded space + path + } + for _, in := range invalid { + if got := sanitizeForwardedHost(in); got != "" { + t.Fatalf("host %q must be rejected (delimiter would poison the redirect base), got %q", in, got) + } + } +} diff --git a/internal/utils/utils.go b/internal/utils/utils.go index b6ddf68..f6c2a0a 100644 --- a/internal/utils/utils.go +++ b/internal/utils/utils.go @@ -137,8 +137,12 @@ func DetermineScheme(req *http.Request, forceHTTPS bool) string { return "https" } - // Check X-Forwarded-Proto header for proxy scenarios - if scheme := req.Header.Get("X-Forwarded-Proto"); scheme != "" { + // Check X-Forwarded-Proto header for proxy scenarios. Only trusted + // values are accepted: a malformed/junk scheme (e.g. an attacker-set + // "X-Forwarded-Proto: ftp") would otherwise be emitted verbatim into + // redirect_uri as a broken or weakened scheme. Invalid values fall + // through to the TLS check (R103). + if scheme := req.Header.Get("X-Forwarded-Proto"); scheme == "https" || scheme == "http" { return scheme } @@ -177,7 +181,17 @@ func sanitizeForwardedHost(v string) string { if v == "" { return "" } - if strings.IndexFunc(v, func(r rune) bool { return r < 0x20 || r == 0x7f || r == ' ' }) >= 0 { + if strings.IndexFunc(v, func(r rune) bool { + // Reject control chars, whitespace, and the delimiters that would + // let a crafted X-Forwarded-Host smuggle a path (''/''), query + // (''?''), fragment (''#''), or userinfo (''@'') into the + // host part of a redirect URL (which is passed to the IdP as + // redirect_uri / post_logout_redirect_uri and matched there). + // Valid host characters — alphanumerics, '-', '.', ':', '[', ']' + // (port / IPv6) — remain allowed. + return r < 0x20 || r == 0x7f || r == ' ' || + r == '/' || r == '?' || r == '#' || r == '@' || r == '\\' + }) >= 0 { return "" } return v diff --git a/internal/utils/utils_url_test.go b/internal/utils/utils_url_test.go index 9713361..06fbcbf 100644 --- a/internal/utils/utils_url_test.go +++ b/internal/utils/utils_url_test.go @@ -9,11 +9,11 @@ import ( // TestDetermineScheme verifies the scheme priority: forceHTTPS > X-Forwarded-Proto > TLS > http. func TestDetermineScheme(t *testing.T) { cases := []struct { - name string - forceHTTPS bool - fwdProto string - tls bool - want string + name string + forceHTTPS bool + fwdProto string + tls bool + want string }{ {"force https wins over header", true, "http", false, "https"}, {"force https wins over tls absence", true, "", false, "https"}, @@ -21,6 +21,11 @@ func TestDetermineScheme(t *testing.T) { {"forwarded proto http", false, "http", true, "http"}, {"tls connection", false, "", true, "https"}, {"default http", false, "", false, "http"}, + // R103: a malformed X-Forwarded-Proto must not be emitted verbatim + // into redirect_uri (would yield a broken/weakened scheme). It + // falls through to the TLS check / http default instead. + {"invalid proto falls through to tls", false, "ftp", true, "https"}, + {"invalid proto falls through to http", false, "ftp", false, "http"}, } for _, c := range cases { t.Run(c.name, func(t *testing.T) { @@ -41,10 +46,10 @@ func TestDetermineScheme(t *testing.T) { // TestDetermineHost verifies forwarding-header preference and fallback to req.Host. func TestDetermineHost(t *testing.T) { cases := []struct { - name string - fwd string - host string - want string + name string + fwd string + host string + want string }{ {"forwarded host used", "api.example.com", "backend.local", "api.example.com"}, {"first of comma list taken", "a.example.com, b.example.com", "backend.local", "a.example.com"}, diff --git a/memory_monitor.go b/memory_monitor.go index 5ec4276..289d5fd 100644 --- a/memory_monitor.go +++ b/memory_monitor.go @@ -427,10 +427,19 @@ func (mm *MemoryMonitor) StartMonitoring(ctx context.Context, interval time.Dura // Check if monitoring is already started if globalMonitoringStarted { - if !isTestMode() { - mm.logger.Debug("Memory monitoring already started, skipping duplicate start") + // If the singleton task was torn down out-of-band (plugin Close -> + // TaskRegistry.StopAllTasks), the process-global flag survives the + // teardown. Detect that and allow restart so memory monitoring + // resumes for a recreated plugin instance; otherwise a config + // reload would permanently disable monitoring. + if !GetResourceManager().IsTaskRunning("memory-monitor") { + globalMonitoringStarted = false + } else { + if !isTestMode() { + mm.logger.Debug("Memory monitoring already started, skipping duplicate start") + } + return } - return } if interval <= 0 { @@ -538,15 +547,19 @@ func (mm *MemoryMonitor) StopMonitoring() { return } - registry := GetGlobalTaskRegistry() - if task, exists := registry.GetTask("memory-monitor"); exists { - task.Stop() - globalMonitoringStarted = false - if !isTestMode() { - mm.logger.Info("Stopped global memory monitoring") - } - } else { - mm.logger.Errorf("Failed to find memory monitoring task to stop") + // The memory-monitor task is a process-global singleton registered via + // CreateSingletonTask, which stores the task in the ResourceManager's + // registry — NOT in TaskRegistry.tasks (R88 split). Look it up there; + // previously GetGlobalTaskRegistry().GetTask read TaskRegistry.tasks, + // always returned not-exists, so Stop was never called, the ticker + // goroutine leaked past shutdown, and this logged a spurious error. + if err := GetResourceManager().StopBackgroundTask("memory-monitor"); err != nil { + mm.logger.Errorf("Failed to find memory monitoring task to stop: %v", err) + return + } + globalMonitoringStarted = false + if !isTestMode() { + mm.logger.Info("Stopped global memory monitoring") } } @@ -580,11 +593,11 @@ func ResetGlobalMemoryMonitor() { defer globalMonitoringMutex.Unlock() if globalMemoryMonitor != nil { - // Stop monitoring if it's active + // Stop monitoring if it's active (singleton task lives in the + // ResourceManager's registry, see StopMonitoring). if globalMonitoringStarted { - registry := GetGlobalTaskRegistry() - if task, exists := registry.GetTask("memory-monitor"); exists { - task.Stop() + if err := GetResourceManager().StopBackgroundTask("memory-monitor"); err != nil { + GetSingletonNoOpLogger().Error("Failed to find memory monitoring task to stop: %v", err) } } globalMemoryMonitor = nil diff --git a/sharded_cache.go b/sharded_cache.go index 093198b..bb1b5e4 100644 --- a/sharded_cache.go +++ b/sharded_cache.go @@ -88,8 +88,17 @@ func (c *ShardedCache) Get(key string) (interface{}, bool) { // Check expiration if !item.expiresAt.IsZero() && time.Now().After(item.expiresAt) { - // Item expired - remove it lazily - c.Delete(key) + // Item expired — remove it lazily, but ONLY if this is still the + // same entry. A concurrent Set/SetIfAbsent that refreshed the key + // (e.g. a freshly-recorded replay JTI) since this read must not be + // removed, else the just-recorded JTI is lost and a duplicate token + // could pass (R129; matches deleteIfExpired in + // internal/cache/backends/memory_shard.go). + shard.mu.Lock() + if cur, ok := shard.items[key]; ok && cur == item { + delete(shard.items, key) + } + shard.mu.Unlock() return nil, false } @@ -107,9 +116,11 @@ func (c *ShardedCache) Set(key string, value interface{}, ttl time.Duration) { } shard.mu.Lock() - // Check if we need to evict items - if len(shard.items) >= c.maxPerShard { - // Simple eviction: remove expired items first, then oldest + // Overwriting an existing key needs no free slot; only evict when + // inserting a genuinely new key at capacity, so an overwrite never + // drops an unrelated oldest still-valid entry and shrinks the shard + // by one (repeats the R57 hazard) (R153). + if _, present := shard.items[key]; !present && len(shard.items) >= c.maxPerShard { c.evictFromShardLocked(shard) } @@ -140,6 +151,7 @@ func (c *ShardedCache) SetIfAbsent(key string, value interface{}, ttl time.Durat shard.mu.Lock() defer shard.mu.Unlock() + _, found := shard.items[key] if it, exists := shard.items[key]; exists { // Present and not expired -> duplicate. if it.expiresAt.IsZero() || !time.Now().After(it.expiresAt) { @@ -148,7 +160,9 @@ func (c *ShardedCache) SetIfAbsent(key string, value interface{}, ttl time.Durat // Present but expired -> treat as absent and replace below. } - if len(shard.items) >= c.maxPerShard { + // The replace-an-expired path needs no free slot; only evict when the + // key is absent and the shard is at capacity (R153). + if !found && len(shard.items) >= c.maxPerShard { c.evictFromShardLocked(shard) } diff --git a/singleton_resources.go b/singleton_resources.go index d4036f7..ab8f05f 100644 --- a/singleton_resources.go +++ b/singleton_resources.go @@ -124,20 +124,25 @@ func (rm *ResourceManager) RegisterBackgroundTask(name string, interval time.Dur defer rm.tasksMu.Unlock() // If a task with this name already exists and is still running, keep it - // (idempotent re-registration from another middleware instance). + // (idempotent re-registration from another middleware instance). Keep it + // only when it has STARTED (started==1 && stopped==0). A task whose + // startOnce was consumed without ever starting — e.g. Start was + // rejected by the circuit breaker, or Stop raced re-registration — + // would otherwise be kept here, then StartBackgroundTask would be a + // permanent no-op (Start is sync.Once-guarded) and the singleton + // task would never run again until process restart. if existing, exists := rm.tasks[name]; exists { - if atomic.LoadInt32(&existing.stopped) == 0 { + if atomic.LoadInt32(&existing.started) == 1 && atomic.LoadInt32(&existing.stopped) == 0 { if rm.logger != nil { rm.logger.Debugf("Background task %s already registered", name) } return nil } - // The existing task was stopped. Reusing the same BackgroundTask - // object would leave it permanently dead: BackgroundTask.Start is - // guarded by sync.Once, so once stopped it can never restart, yet - // IsTaskRunning returns false and callers will call Start again. - // Replace it with a fresh task so re-registration after a - // close+recreate cycle (same provider URL) actually resumes it. + // The existing task was either stopped or never actually started. + // Reusing the same BackgroundTask object would leave it permanently + // dead: BackgroundTask.Start is guarded by sync.Once, so once + // consumed it can never start again. Replace it with a fresh task + // so re-registration actually resumes it. } task := NewBackgroundTask(name, interval, taskFunc, rm.logger, &rm.wg) diff --git a/universal_cache.go b/universal_cache.go index 624d2ba..cc20255 100644 --- a/universal_cache.go +++ b/universal_cache.go @@ -16,11 +16,13 @@ import ( type CacheType string const ( - CacheTypeToken CacheType = "token" - CacheTypeMetadata CacheType = "metadata" - CacheTypeJWK CacheType = "jwk" - CacheTypeSession CacheType = "session" - CacheTypeGeneral CacheType = "general" + CacheTypeToken CacheType = "token" + CacheTypeBlacklist CacheType = "blacklist" + CacheTypeMetadata CacheType = "metadata" + CacheTypeIntrospection CacheType = "introspection" + CacheTypeJWK CacheType = "jwk" + CacheTypeSession CacheType = "session" + CacheTypeGeneral CacheType = "general" // maxCacheEntrySize defines the maximum size for a single cache entry (64 MiB) // This prevents integer overflow when allocating memory for serialization @@ -248,7 +250,17 @@ func (c *UniversalCache) Set(key string, value interface{}, ttl time.Duration) e if err := c.backend.Set(ctx, c.prefixKey(key), data, ttl); err != nil { c.logger.Infof("Backend set error for key %s: %v", key, err) - // Continue with local cache even if backend fails + // The backend still holds an OLDER entry for this key. If we + // leave it in place, a subsequent Get treats the backend as + // authoritative and (via updateLocalCache) overwrites the + // just-written local value with the stale one (R162). Evict the + // stale backend entry so Get falls through to the fresh local + // value instead of resurrecting the old one. + dctx, dcancel := context.WithTimeout(context.Background(), 200*time.Millisecond) + if _, derr := c.backend.Delete(dctx, c.prefixKey(key)); derr != nil { + c.logger.Debugf("Backend delete after failed set for key %s: %v", key, derr) + } + dcancel() } } @@ -294,8 +306,12 @@ func (c *UniversalCache) setLocal(key string, value interface{}, ttl time.Durati } } - // Check size limits - if c.lruList.Len() >= c.config.MaxSize { + // Check size limits — only evict when inserting a NEW key. An in-place + // update of an existing key adds no net entry, so evicting would drop an + // unrelated live LRU entry and leave the cache one below capacity (cache + // thrashes under sustained traffic at MaxSize). Mirror of the shard + // backend fix (R139 F2). + if _, present := c.items[key]; !present && c.lruList.Len() >= c.config.MaxSize { c.evictOldest() } @@ -397,7 +413,7 @@ func (c *UniversalCache) getLocal(key string) (interface{}, bool) { // previous unconditional Lock serialized every JWT verify on a single // mutex and pinned a CPU under load. switch c.config.Type { - case CacheTypeToken, CacheTypeJWK, CacheTypeSession: + case CacheTypeToken, CacheTypeJWK, CacheTypeSession, CacheTypeIntrospection: c.mu.RLock() item, exists := c.items[key] if !exists { @@ -975,8 +991,11 @@ func (c *UniversalCache) updateLocalCache(key string, value interface{}, ttl tim } } - // Check size limits - if c.lruList.Len() >= c.config.MaxSize { + // Check size limits — only evict when inserting a NEW key. An in-place + // update adds no net entry (the replace below frees the old size), so + // evicting would drop an unrelated live LRU entry and leave the cache + // one below capacity. See setLocal (R141). + if _, present := c.items[key]; !present && c.lruList.Len() >= c.config.MaxSize { c.evictOldest() } diff --git a/universal_cache_singleton.go b/universal_cache_singleton.go index 0081b8b..6461b3f 100644 --- a/universal_cache_singleton.go +++ b/universal_cache_singleton.go @@ -94,6 +94,23 @@ func GetUniversalCacheManagerWithConfig(logger *Logger, redisConfig *RedisConfig return universalCacheManager } +// newBlacklistCacheConfig returns the UniversalCacheConfig used for the +// shared token-blacklist cache. It MUST keep a CacheType distinct from +// the token cache's CacheTypeToken: prefixKey composes the backend key +// from c.config.Type, so a shared type would put blacklist and token +// entries for the same raw token string in the same (e.g. Redis) +// namespace — both rejecting valid claims-cached tokens as blacklisted +// and letting token caching clobber a real blacklist marker (R128). +func newBlacklistCacheConfig(logger *Logger) UniversalCacheConfig { + return UniversalCacheConfig{ + Type: CacheTypeBlacklist, + MaxSize: 1000, + DefaultTTL: 24 * time.Hour, + Logger: logger, + SkipAutoCleanup: true, // Managed cleanup + } +} + // initializeDefaultCaches initializes caches with memory-only backends func initializeDefaultCaches(manager *UniversalCacheManager, logger *Logger) { // Initialize token cache - CRITICAL FIX: Reduced from 5000 to 1000 @@ -106,14 +123,12 @@ func initializeDefaultCaches(manager *UniversalCacheManager, logger *Logger) { SkipAutoCleanup: true, // Managed cleanup }) - // Initialize blacklist cache - manager.blacklistCache = NewUniversalCache(UniversalCacheConfig{ - Type: CacheTypeToken, - MaxSize: 1000, - DefaultTTL: 24 * time.Hour, - Logger: logger, - SkipAutoCleanup: true, // Managed cleanup - }) + // Initialize blacklist cache. Distinct CacheType so its (possibly + // shared-backend) "blacklist:" namespace can never collide with the + // token cache's "token:" namespace for the same raw token string; a + // collision would both reject valid claims-cached tokens as blacklisted + // and let token caching clobber a real blacklist marker (R128). + manager.blacklistCache = NewUniversalCache(newBlacklistCacheConfig(logger)) // Initialize metadata cache with grace periods manager.metadataCache = NewUniversalCache(UniversalCacheConfig{ @@ -156,8 +171,13 @@ func initializeDefaultCaches(manager *UniversalCacheManager, logger *Logger) { }) // Initialize introspection cache for OAuth 2.0 Token Introspection (RFC 7662) + // Uses its own CacheType so its Redis namespace differs from the blacklist + // (both in CacheTypeToken): with a shared Redis backend an opaque token is keyed + // by its raw value in both caches, so sharing the "token:" prefix let an + // introspection write clobber a same-token blacklist marker (revoked opaque + // token could be re-accepted) across replicas. See R114. manager.introspectionCache = NewUniversalCache(UniversalCacheConfig{ - Type: CacheTypeToken, // Use token cache type for introspection results + Type: CacheTypeIntrospection, MaxSize: 1000, // Cache up to 1000 introspection results DefaultTTL: 5 * time.Minute, // Short TTL for security (introspect frequently) Logger: logger, @@ -297,15 +317,13 @@ func initializeCachesWithRedis(manager *UniversalCacheManager, logger *Logger, r createBackend(CacheTypeToken), ) - // Initialize blacklist cache (CRITICAL - must be consistent across replicas) + // Initialize blacklist cache (CRITICAL - must be consistent across replicas). + // Distinct CacheType so its "blacklist:" Redis namespace can't collide with + // the token cache's "token:" namespace for the same raw token (R128); a + // shared prefix would reject valid claims-cached tokens as blacklisted or + // let token caching clobber a real blacklist marker. manager.blacklistCache = NewUniversalCacheWithBackend( - UniversalCacheConfig{ - Type: CacheTypeToken, - MaxSize: 1000, - DefaultTTL: 24 * time.Hour, - Logger: logger, - SkipAutoCleanup: true, // Managed cleanup - }, + newBlacklistCacheConfig(logger), createBackend("blacklist"), ) @@ -355,16 +373,18 @@ func initializeCachesWithRedis(manager *UniversalCacheManager, logger *Logger, r SkipAutoCleanup: true, // Managed cleanup }) - // Introspection cache uses backend for sharing results + // Introspection cache uses backend for sharing results. Distinct CacheType + // so its Redis "introspection:" namespace can't collide with the blacklist's + // "token:" namespace for the same raw opaque token (R114). manager.introspectionCache = NewUniversalCacheWithBackend( UniversalCacheConfig{ - Type: CacheTypeToken, + Type: CacheTypeIntrospection, MaxSize: 1000, DefaultTTL: 5 * time.Minute, Logger: logger, SkipAutoCleanup: true, // Managed cleanup }, - createBackend(CacheTypeToken), + createBackend(CacheTypeIntrospection), ) // Token type cache stays memory-only (local optimization) diff --git a/utilities.go b/utilities.go index 6e87d8e..d833fee 100644 --- a/utilities.go +++ b/utilities.go @@ -10,6 +10,7 @@ import ( "html" "net/http" "runtime" + "strconv" "strings" "time" ) @@ -136,9 +137,7 @@ func (t *TraefikOidc) isAllowedDomain(email string) bool { } domain := strings.ToLower(parts[1]) - _, domainAllowed := t.allowedUserDomains[domain] - - if domainAllowed { + if t.domainAllowed(domain) { t.logger.Debugf("Email domain %s is allowed", domain) return true } else { @@ -153,6 +152,97 @@ func (t *TraefikOidc) isAllowedDomain(email string) bool { return false } +// domainAllowed reports whether domain is an allowed user domain, +// including subdomains thereof. An operator configuring +// "example.com" intends to cover staff on corporate subdomains, so +// "mail.example.com" (and deeper) must be accepted; exact match is +// no longer the only path. A plain suffix lookup is avoided (would +// also match "badexample.com"), so subdomains require a leading "." +// separator. (R152) +func (t *TraefikOidc) domainAllowed(domain string) bool { + if _, ok := t.allowedUserDomains[domain]; ok { + return true + } + for allowed := range t.allowedUserDomains { + if len(domain) > len(allowed) && strings.HasSuffix(domain, "."+allowed) { + return true + } + } + return false +} + +// emailVerifiedTrue extracts the boolean email_verified claim value, +// claimScalarString returns the string form of a scalar claim value, +// coercing JSON numbers (float64, json.Number, int variants) to their +// decimal string representation so a numeric identifier claim (e.g. a +// numeric sub from a non-conformant IdP) is preserved rather than +// silently dropped. second is false for non-scalar values (maps, +// slices, objects) so callers can treat them as invalid. +func claimScalarString(raw interface{}) (s string, ok bool) { + switch v := raw.(type) { + case string: + return v, true + case json.Number: + return v.String(), true + case float64: + return strconv.FormatFloat(v, 'f', -1, 64), true + case float32: + return strconv.FormatFloat(float64(v), 'f', -1, 32), true + case int: + return strconv.Itoa(v), true + case int64: + return strconv.FormatInt(v, 10), true + default: + return "", false + } +} + +// emailVerifiedTrue extracts the boolean email_verified claim value, +// tolerating bool and string representations. present is false when the +// claim is absent or of an uninterpretable type (the caller then decides +// how to treat absence). +func emailVerifiedTrue(v interface{}) (verified, present bool) { + switch val := v.(type) { + case bool: + return val, true + case string: + switch strings.ToLower(strings.TrimSpace(val)) { + case "true", "1": + return true, true + case "false", "0": + return false, true + } + case float64: + return val != 0, true + case json.Number: + f, err := val.Float64() + if err != nil { + return false, false + } + return f != 0, true + } + return false, false +} + +// emailIdentityPermitted reports whether an email-based user identifier may be +// admitted for email/domain authorization given the ID-token claims. An email +// is admitted unless the IdP explicitly marks it unverified +// (email_verified == false): such an address must not satisfy an email +// allowlist, since the user may not control it (impersonation within an +// allowed domain). Absence is tolerated for backward compatibility (the +// middleware and its test fixtures have long run without the claim; some IdPs +// do not emit it). This matches the bearer-path stance that untrusted email +// is a spoofing vector (main.go). +func emailIdentityPermitted(userIdentifier string, claims map[string]interface{}) bool { + if !strings.Contains(userIdentifier, "@") { + return true + } + if verified, present := emailVerifiedTrue(claims["email_verified"]); present && !verified { + return false + } + return true +} + // keysFromMap extracts string keys from a map for logging purposes. // Helper function to get keys from a map for logging. // Parameters: @@ -176,6 +266,10 @@ func keysFromMap(m map[string]struct{}) []string { // - message: The error message to display. // - code: The HTTP status code to set for the response. func (t *TraefikOidc) sendErrorResponse(rw http.ResponseWriter, req *http.Request, message string, code int) { + // Auth-failure responses (401/403/429/503) must not be cached by + // browsers or intermediaries — a cached 401 body would persist across + // re-authentication (R101). + rw.Header().Set("Cache-Control", "no-store") acceptHeader := req.Header.Get("Accept") if strings.Contains(acceptHeader, "application/json") {