Fix 6 issues surfaced by adversarial agent team (round 35)

Concurrency/lifecycle (P1, highest-value):
- singleton_resources.go: RegisterBackgroundTask now REPLACES a stopped
  task instead of reusing it. BackgroundTask.Start is sync.Once-guarded,
  so a task stopped on instance close and re-registered (Traefik
  close+recreate for the same provider) was permanently dead: IsTaskRunning
  returned false (started && !stopped) so callers retried a start that
  was a no-op. Metadata refresh, token cleanup, and health checks now
  resume across a recreate cycle.
- error_recovery.go: the shared graceful-degradation health task was
  bound to the FIRST instance's performHealthChecks (later instances'
  checks never ran) and any instance's Close stopped it for all. Added a
  global live-instance registry; the shared task runs every instance's
  checks and is stopped only when the last instance closes (mirrors the
  singleton-token-cleanup gate).

Storage/cache:
- hybrid.go recordL2Error: Store(now) then immediately compared
  time.Since(t)<1s against its own write => ALWAYS fired, so a single
  transient L2 error flipped the whole cache into fallback (full L2
  bypass -> upstream stampede). Now enters fallback only on a PRIOR
  error within the window (true "multiple errors"). Updated two tests
  that had encoded the old single-error-fallback bug.
- redis_pool.go: ConnectionPool.Put checked closed but sent on
  p.connections without holding p.mu; Close closed it under p.mu.
  Window => 'send on closed channel' panic at shutdown. Put now holds
  p.mu (the lock only Close used).

HTTP/transport:
- jwk.go fetchJWKS: added an explicit >1 MiB guard (was silent partial
  read, could yield a truncated keyset and spurious missing-kid).
- http_client_pool.go configKey: now includes every transport-behavior
  field it actually consumes (compression, HTTP2, keep-alives, timeouts,
  buffer sizes, dial) so behaviorally different configs no longer share
  one transport built with the first config's settings.

Each fix has a regression test. Verified: full source-order suite green,
golangci-lint 0 issues, 3x full -race -count=1 -shuffle=on ./... all exit 0.

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
+		// Stop health check task only when this was the last live instance.
-		gd.mutex.Lock()
+		// The task is process-global and shared by every GracefulDegradation
-		task := gd.healthCheckTask
+		// instance; stopping it on any one instance's close would kill
-		gd.mutex.Unlock()
+		// 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 {
+		if lastInstance {
-			task.Stop()
+			gd.mutex.Lock()
-			// Don't set to nil to avoid race conditions
+			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")

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
+	// Pool transports by every parameter the built *http.Transport actually
-	// behavior. RootCAs and InsecureSkipVerify MUST be part of the key:
+	// 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,
 	)
 }
 

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

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")
+}

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)

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()
+}

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 {

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).

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 a task with this name already exists and is still running, keep it
-	if _, exists := rm.tasks[name]; exists {
+	// (idempotent re-registration from another middleware instance).
-		if rm.logger != nil {
+	if existing, exists := rm.tasks[name]; exists {
-			rm.logger.Debugf("Background task %s already registered", name)
+		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
+		// The existing task was stopped. Reusing the same BackgroundTask
-		return nil
+		// 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