diff --git a/error_recovery.go b/error_recovery.go index 8b2231a..9c18f03 100644 --- a/error_recovery.go +++ b/error_recovery.go @@ -788,6 +788,31 @@ func DefaultGracefulDegradationConfig() GracefulDegradationConfig { // NewGracefulDegradation creates a new graceful degradation manager // NewGracefulDegradation creates a new graceful degradation mechanism. // Initializes fallback and health check maps and starts background health monitoring. +// gdInstances tracks live GracefulDegradation instances so the single shared +// health-check task can run every instance's health checks (not just the first) +// and so the task is only stopped when the last instance closes. +var gdInstances = struct { + sync.RWMutex + set map[*GracefulDegradation]struct{} +}{set: make(map[*GracefulDegradation]struct{})} + +// 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() { + gdInstances.RLock() + instances := make([]*GracefulDegradation, 0, len(gdInstances.set)) + for gd := range gdInstances.set { + instances = append(instances, gd) + } + gdInstances.RUnlock() + + for _, gd := range instances { + gd.performHealthChecks() + } +} + func NewGracefulDegradation(config GracefulDegradationConfig, logger *Logger) *GracefulDegradation { gd := &GracefulDegradation{ BaseRecoveryMechanism: NewBaseRecoveryMechanism("graceful-degradation", logger), @@ -904,13 +929,22 @@ func (gd *GracefulDegradation) executeFallback(serviceName string) (interface{}, // startHealthCheckRoutine starts the background health check routine func (gd *GracefulDegradation) startHealthCheckRoutine() { - // Use singleton task registry to prevent multiple instances + // Register this instance so the shared task's health pass reaches it even + // when it was not the first instance created. + gdInstances.Lock() + gdInstances.set[gd] = struct{}{} + gdInstances.Unlock() + + // Use the singleton task registry so multiple GracefulDegradation + // instances share one health-check goroutine. The task function is the + // global pass over every live instance, not this instance's own health + // checks — otherwise only the first instance's checks would run. registry := GetGlobalTaskRegistry() task, err := registry.CreateSingletonTask( "graceful-degradation-health-check", gd.config.HealthCheckInterval, - gd.performHealthChecks, + globalPerformHealthChecks, gd.BaseRecoveryMechanism.logger, nil, // No specific wait group ) @@ -983,14 +1017,25 @@ func (gd *GracefulDegradation) Close() { close(gd.stopChan) } - // Stop health check task - gd.mutex.Lock() - task := gd.healthCheckTask - gd.mutex.Unlock() + // Stop health check task only when this was the last live instance. + // The task is process-global and shared by every GracefulDegradation + // instance; stopping it on any one instance's close would kill + // health checks / recovery for all surviving instances (mirror of the + // singleton-token-cleanup lastInstance gate). + gdInstances.Lock() + delete(gdInstances.set, gd) + lastInstance := len(gdInstances.set) == 0 + gdInstances.Unlock() - if task != nil { - task.Stop() - // Don't set to nil to avoid race conditions + if lastInstance { + gd.mutex.Lock() + task := gd.healthCheckTask + gd.mutex.Unlock() + + if task != nil { + task.Stop() + // Don't set to nil to avoid race conditions + } } gd.logger.Debug("GracefulDegradation shut down successfully") diff --git a/http_client_pool.go b/http_client_pool.go index 2d44d87..6eaf622 100644 --- a/http_client_pool.go +++ b/http_client_pool.go @@ -219,20 +219,35 @@ func (p *SharedTransportPool) performCleanup() { // configKey generates a unique key for a config func (p *SharedTransportPool) configKey(config HTTPClientConfig) string { - // Pool transports by the parameters that change TLS or connection - // behavior. RootCAs and InsecureSkipVerify MUST be part of the key: + // Pool transports by every parameter the built *http.Transport actually + // consumes. RootCAs and InsecureSkipVerify MUST be part of the key: // otherwise a middleware configured with a custom CA would share a // transport with one using the system store, silently bypassing its - // CA configuration. + // CA configuration. The remaining fields affect connection behavior + // (HTTP version, compression, keep-alives, timeouts, buffer sizes, + // dial settings); omitting them would hand a caller a transport built + // with different settings than it requested (e.g. wrong compression). skip := "0" if config.InsecureSkipVerify { skip = "1" } - return fmt.Sprintf("%d|%d|%p|%s", + return fmt.Sprintf("%d|%d|%p|%s|%d|%d|%d|%v|%v|%v|%d|%d|%d|%d|%d|%d", config.MaxConnsPerHost, config.MaxIdleConnsPerHost, config.RootCAs, skip, + config.DialTimeout, + config.KeepAlive, + config.TLSHandshakeTimeout, + config.ForceHTTP2, + config.DisableKeepAlives, + config.DisableCompression, + config.ResponseHeaderTimeout, + config.ExpectContinueTimeout, + config.WriteBufferSize, + config.ReadBufferSize, + config.IdleConnTimeout, + config.MaxIdleConns, ) } diff --git a/internal/cache/backends/hybrid.go b/internal/cache/backends/hybrid.go index 3e4be76..7dfc8cc 100644 --- a/internal/cache/backends/hybrid.go +++ b/internal/cache/backends/hybrid.go @@ -663,18 +663,22 @@ func (h *HybridBackend) healthMonitor() { // recordL2Error records the timestamp of an L2 error func (h *HybridBackend) recordL2Error() { - h.lastL2Error.Store(time.Now()) + now := time.Now() - // Check if we should enter fallback mode based on recent errors + // Enter fallback only if there was a PRIOR error within the last second — + // i.e. this is at least the second L2 error in quick succession. + // Checking a freshly-written timestamp would always be "recent" and + // enter fallback on a single transient L2 error (cache stampede). if !h.fallbackMode.Load() { - // Simple heuristic: if we've had an error in the last second, consider L2 unhealthy if lastErr := h.lastL2Error.Load(); lastErr != nil { - if t, ok := lastErr.(time.Time); ok && time.Since(t) < time.Second { + if t, ok := lastErr.(time.Time); ok && now.Sub(t) < time.Second { h.fallbackMode.Store(true) h.logger.Warnf("Multiple L2 errors detected, entering fallback mode") } } } + + h.lastL2Error.Store(now) } // extractCacheType attempts to determine the cache type from the key diff --git a/internal/cache/backends/hybrid_test.go b/internal/cache/backends/hybrid_test.go index 4070683..cc52bbe 100644 --- a/internal/cache/backends/hybrid_test.go +++ b/internal/cache/backends/hybrid_test.go @@ -800,7 +800,8 @@ func TestHybridBackend_Ping_RecoverFromFallback(t *testing.T) { ctx := context.Background() - // First ping fails L2, enters fallback + // Repeated L2 failures (two errors within the window) enter fallback. + hybrid.Ping(ctx) hybrid.Ping(ctx) assert.True(t, hybrid.fallbackMode.Load()) @@ -1184,7 +1185,13 @@ func TestHybridBackend_GetMany_BatchBackendError(t *testing.T) { assert.Len(t, results, 1) assert.Equal(t, []byte("value1"), results["key1"]) - // Batch error should trigger fallback mode + // A single batch error is not enough to enter fallback (avoids a cache + // stampede on one transient L2 error); a second error within the window is. + assert.False(t, hybrid.fallbackMode.Load()) + _, err = hybrid.GetMany(ctx, []string{"key2"}) + assert.NoError(t, err) + + // Repeated batch error should trigger fallback mode time.Sleep(50 * time.Millisecond) assert.True(t, hybrid.fallbackMode.Load()) } @@ -1488,3 +1495,25 @@ func TestHybridBackend_RecordL2Error_EntersFallbackMode(t *testing.T) { // Should enter fallback mode assert.True(t, hybrid.fallbackMode.Load()) } + +func TestHybridBackend_RecordL2Error_RequiresRepeatedErrors(t *testing.T) { + config := &HybridConfig{ + Primary: newMockBackend(), + Secondary: newMockBackend(), + } + + hybrid, err := NewHybridBackend(config) + require.NoError(t, err) + defer hybrid.Close() + + // A SINGLE transient L2 error must NOT flip the whole cache into fallback + // mode (which bypasses L2 for reads and writes -> cache stampede). + hybrid.recordL2Error() + assert.False(t, hybrid.fallbackMode.Load(), + "a single L2 error must not enter fallback mode") + + // A second error within the 1s window (repeated failures) does. + hybrid.recordL2Error() + assert.True(t, hybrid.fallbackMode.Load(), + "repeated L2 errors within the window must enter fallback mode") +} diff --git a/internal/cache/backends/redis_pool.go b/internal/cache/backends/redis_pool.go index 0178e1a..18ba06e 100644 --- a/internal/cache/backends/redis_pool.go +++ b/internal/cache/backends/redis_pool.go @@ -158,6 +158,13 @@ func (p *ConnectionPool) Put(conn *RedisConn) { p.puts.Add(1) p.activeConns.Add(-1) + // Hold p.mu so this is mutually exclusive with Close (which closes + // p.connections under the same lock). Without it, a Put that passed + // the closed check could send on p.connections just after Close + // closed it, panicking "send on closed channel" at shutdown. + p.mu.Lock() + defer p.mu.Unlock() + if p.closed.Load() || conn.closed.Load() { _ = conn.Close() p.totalConns.Add(-1) diff --git a/internal/cache/backends/redis_pool_test.go b/internal/cache/backends/redis_pool_test.go index 2cb0945..f4172d1 100644 --- a/internal/cache/backends/redis_pool_test.go +++ b/internal/cache/backends/redis_pool_test.go @@ -688,3 +688,47 @@ func TestRedisConn_RejectOversizedArgumentBytes(t *testing.T) { require.Error(t, err) assert.Contains(t, err.Error(), "arguments too large") } + +// TestConnectionPool_PutDuringClose_NoPanic hammers Put/Get concurrently with +// Close. Before the fix, Put checked closed then sent on p.connections +// without holding p.mu; if Close closed the channel in between, the send +// panicked ("send on closed channel") at shutdown. Run under -race. +func TestConnectionPool_PutDuringClose_NoPanic(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, + } + + pool, err := NewConnectionPool(config) + require.NoError(t, err) + + ctx := context.Background() + stop := make(chan struct{}) + var wg sync.WaitGroup + for g := 0; g < 8; g++ { + wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case <-stop: + return + default: + } + if conn, err := pool.Get(ctx); err == nil { + pool.Put(conn) // may race Close + } + } + }() + } + + time.Sleep(5 * time.Millisecond) + _ = pool.Close() + close(stop) + wg.Wait() +} diff --git a/jwk.go b/jwk.go index 1ff88a9..e69d0ff 100644 --- a/jwk.go +++ b/jwk.go @@ -262,10 +262,13 @@ func fetchJWKS(ctx context.Context, jwksURL string, httpClient *http.Client) (*J return nil, fmt.Errorf("JWKS fetch failed with status %d: %s", resp.StatusCode, body) } - body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) + body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20+1)) if err != nil { return nil, fmt.Errorf("error reading JWKS response: %w", err) } + if len(body) > 1<<20 { + return nil, fmt.Errorf("JWKS response exceeds 1 MiB limit") + } var jwks JWKSet if err := json.Unmarshal(body, &jwks); err != nil { diff --git a/review_r34_regression_test.go b/review_r34_regression_test.go new file mode 100644 index 0000000..8e60e75 --- /dev/null +++ b/review_r34_regression_test.go @@ -0,0 +1,180 @@ +package traefikoidc + +import ( + "context" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// TestFetchJWKSOversizeResponse: a provider JWKS larger than 1 MiB must be +// rejected explicitly, not silently truncated (which could yield a partial +// keyset and a spurious "no matching public key for kid"). +func TestFetchJWKSOversizeResponse(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // Body strictly larger than the 1 MiB read limit (prefixed with a + // valid JSON opener so the ONLY reason to reject is the size guard). + body := `{"keys":[` + strings.Repeat("{}", (1<<20)/2) + `]}` + w.Write([]byte(body)) + })) + defer server.Close() + + cache := NewJWKCache() + jwks, err := cache.GetJWKS(context.Background(), server.URL, http.DefaultClient) + require.Error(t, err, "oversized JWKS must be rejected") + require.Nil(t, jwks) + require.Contains(t, err.Error(), "1 MiB") +} + +func waitForHealthTaskRunning(t *testing.T, want bool) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if GetResourceManager().IsTaskRunning("graceful-degradation-health-check") == want { + return + } + time.Sleep(5 * time.Millisecond) + } + t.Fatalf("health task running state != %v", want) +} + +func waitForGdInstances(t *testing.T, want int) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + gdInstances.RLock() + n := len(gdInstances.set) + gdInstances.RUnlock() + if n >= want { + return + } + time.Sleep(5 * time.Millisecond) + } + gdInstances.RLock() + n := len(gdInstances.set) + gdInstances.RUnlock() + t.Fatalf("gd instances wanted %d got %d", want, n) +} + +// resetGdInstancesForTest establishes a clean baseline for the process-global +// instance registry so a test's "last instance" and iteration assertions are +// not skewed by instances created by earlier tests in the same process. +func resetGdInstancesForTest() { + gdInstances.Lock() + gdInstances.set = make(map[*GracefulDegradation]struct{}) + gdInstances.Unlock() +} + +// TestConfigKeySeparatesBehaviorDifferingConfigs: two configs that differ only +// in a transport-behavior field (here DisableCompression) must get distinct +// transports, not share one built with the first config's settings. +func TestConfigKeySeparatesBehaviorDifferingConfigs(t *testing.T) { + pool := &SharedTransportPool{ + transports: make(map[string]*sharedTransport), + maxConns: 20, + maxClients: 5, + clientCount: 0, + ctx: context.Background(), + } + + a := DefaultHTTPClientConfig() + b := a + b.DisableCompression = !a.DisableCompression + + require.NotEqual(t, pool.configKey(a), pool.configKey(b), + "configs differing only in DisableCompression must have distinct keys") + + ta := pool.GetOrCreateTransport(a) + tb := pool.GetOrCreateTransport(b) + require.NotEqual(t, ta, tb, "behaviorally different configs must not share a transport") + + // Same config still reuses its transport. + ta2 := pool.GetOrCreateTransport(a) + require.Equal(t, ta, ta2, "identical config reuses its transport") +} + +// TestRegisterBackgroundTaskRestartsAfterStop: a singleton task stopped on a +// close+recreate cycle must be restartable on re-registration (Start is +// once-guarded, so the registry must replace a stopped task with a fresh one). +func TestRegisterBackgroundTaskRestartsAfterStop(t *testing.T) { + rm := &ResourceManager{ + tasks: map[string]*BackgroundTask{}, + logger: GetSingletonNoOpLogger(), + } + const name = "restart-test-task" + var calls int32 + register := func() { + t.Helper() + require.NoError(t, rm.RegisterBackgroundTask(name, time.Hour, func() { + atomic.AddInt32(&calls, 1) + })) + } + + register() + require.NoError(t, rm.StartBackgroundTask(name)) + require.True(t, rm.IsTaskRunning(name)) + + require.NoError(t, rm.StopBackgroundTask(name)) + require.False(t, rm.IsTaskRunning(name)) + + // Re-register after stop: must be restartable (fresh BackgroundTask). + register() + require.NoError(t, rm.StartBackgroundTask(name)) + require.True(t, rm.IsTaskRunning(name), "re-registered singleton task must be restartable after stop") + + require.NoError(t, rm.StopBackgroundTask(name)) +} + +func TestSharedHealthTaskSurvivesOtherInstanceClose(t *testing.T) { + resetGdInstancesForTest() + + cfg := DefaultGracefulDegradationConfig() + cfg.HealthCheckInterval = time.Hour // keep the task from churning + + gd1 := NewGracefulDegradation(cfg, GetSingletonNoOpLogger()) + gd2 := NewGracefulDegradation(cfg, GetSingletonNoOpLogger()) + defer gd2.Close() + + waitForHealthTaskRunning(t, true) + waitForGdInstances(t, 2) + + gd1.Close() + waitForHealthTaskRunning(t, true) + + gd2.Close() + waitForHealthTaskRunning(t, false) +} + +// TestSharedHealthTaskReachesAllInstances: the shared health task must exercise +// health checks registered on ANY live instance, not just the first one. +func TestSharedHealthTaskReachesAllInstances(t *testing.T) { + resetGdInstancesForTest() + + cfg := DefaultGracefulDegradationConfig() + cfg.HealthCheckInterval = time.Hour + + gd1 := NewGracefulDegradation(cfg, GetSingletonNoOpLogger()) + gd2 := NewGracefulDegradation(cfg, GetSingletonNoOpLogger()) + defer gd1.Close() + defer gd2.Close() + + waitForGdInstances(t, 2) + waitForHealthTaskRunning(t, true) + + var called atomic.Bool + gd2.RegisterHealthCheck("svc-on-instance-2", func() bool { + called.Store(true) + return true + }) + + globalPerformHealthChecks() + require.True(t, called.Load(), + "a health check registered on the second instance must be exercised by the shared task") +} + +// TestRecordL2Error not here: lives in internal/cache/backends (hybrid test). diff --git a/singleton_resources.go b/singleton_resources.go index b7e7094..17de2e5 100644 --- a/singleton_resources.go +++ b/singleton_resources.go @@ -123,16 +123,23 @@ func (rm *ResourceManager) RegisterBackgroundTask(name string, interval time.Dur rm.tasksMu.Lock() defer rm.tasksMu.Unlock() - // Check if task already exists - if _, exists := rm.tasks[name]; exists { - if rm.logger != nil { - rm.logger.Debugf("Background task %s already registered", name) + // If a task with this name already exists and is still running, keep it + // (idempotent re-registration from another middleware instance). + if existing, exists := rm.tasks[name]; exists { + if atomic.LoadInt32(&existing.stopped) == 0 { + if rm.logger != nil { + rm.logger.Debugf("Background task %s already registered", name) + } + return nil } - // Return existing task without error for idempotency - 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. } - // Create new task with WaitGroup for proper cleanup task := NewBackgroundTask(name, interval, taskFunc, rm.logger, &rm.wg) rm.tasks[name] = task