fix(refresh-coordinator): replace global RWMutex with sync.Map

Under Yaegi, the RefreshCoordinator.refreshMutex was held for tens of
milliseconds per request because every operation inside the critical
section (map access, isInCooldown, recordRefreshAttempt,
isUnderMemoryPressure, atomic ops, struct allocation) is dispatched
through reflect.Value.Call with full arg boxing/unboxing.

Concurrent refreshes on the same coordinator serialized into a queue
that grew without bound. Live capture in production (3 Grafana
dashboards left open) showed:
  * 63 goroutines stuck on rc.refreshMutex.Lock() for 1-11 minutes
  * pod pinned at 1000m CPU (GOMAXPROCS=1)
  * 5.15M allocs/sec, 0.45 RPS effective throughput
  * yaegi.call.func9 accounting for 92.66% of cumulative allocs
  * mutex profile dominated by sync.(*Mutex).Unlock via the request chain

Change inFlightRefreshes from map[string]*refreshOperation+RWMutex to
sync.Map and rewrite getOrCreateOperation to:
  1. Speculatively allocate the candidate operation.
  2. Atomically LoadOrStore by tokenHash. Joiners take the existing
     operation; leader takes the new one. No global lock acquired.
  3. Leader runs rate-limit / cooldown / memory-pressure gates AFTER
     the atomic store. Joiners share the leader's outcome via op.done.
  4. Reserve the concurrent-refresh slot via CompareAndSwap so the
     count cannot overshoot in absence of the old serializing lock.
  5. On any gate failure the leader calls failCandidate, which deletes
     the entry from sync.Map, records the error on op.result and closes
     op.done so any joiner that snuck in returns the same error.

performCleanup becomes a single sync.Map.LoadAndDelete, eliminating
the lock entirely on the cleanup path.

Net effect: critical section is no longer Yaegi-interpreted; it
collapses to atomic instructions on a sharded sync.Map. Refresh
contention disappears even under Yaegi.

All tests pass with -race; golangci-lint clean.

README.md

@@ -411,6 +411,19 @@ namespaced claims, Cognito regions, GitLab self-hosted) live in
 
 Set `logLevel: debug` to surface detail.
 
+## Telemetry
+
+On first plugin instantiation this middleware sends a single anonymous
+adoption ping — project name, version, timestamp; no identifiers, no
+request data, no token contents. Fire-and-forget with a 2-second timeout;
+cannot block plugin load or panic.
+
+Local source: [`telemetry.go`](./telemetry.go). Disclosure mirrors
+**[oss-telemetry — Disabling telemetry](https://github.com/lukaszraczylo/oss-telemetry#disabling-telemetry)**.
+
+Quick opt-out: set any of `DO_NOT_TRACK=1`, `OSS_TELEMETRY_DISABLED=1`,
+or `TRAEFIKOIDC_DISABLE_TELEMETRY=1`.
+
 ## License
 
 See [LICENSE](LICENSE).

main.go

@@ -89,6 +89,7 @@ var defaultExcludedURLs = map[string]struct{}{
 //   - The configured TraefikOidc handler ready to process requests.
 //   - An error if essential configuration is missing or invalid (e.g., short encryption key).
 func New(ctx context.Context, next http.Handler, config *Config, name string) (http.Handler, error) {
+	sendTelemetry(pluginVersion)
 	return NewWithContext(ctx, config, next, name)
 }
 

refresh_coordinator.go

@@ -15,7 +15,13 @@ import (
 // It implements request coalescing, rate limiting, and circuit breaking
 // specifically for token refresh operations.
 type RefreshCoordinator struct {
-	inFlightRefreshes      map[string]*refreshOperation
+	// inFlightRefreshes maps tokenHash -> *refreshOperation. sync.Map is used
+	// instead of a plain map + RWMutex so concurrent refreshes do not
+	// serialize on a single global lock. Under Yaegi the previous
+	// refreshMutex.Lock() was held for tens of milliseconds per request due
+	// to interpreter overhead on the work inside the critical section,
+	// causing dozens of goroutines to stack up on it and pin one CPU core.
+	inFlightRefreshes      sync.Map
 	cleanupTimers          map[string]*time.Timer
 	sessionRefreshAttempts map[string]*refreshAttemptTracker
 	circuitBreaker         *RefreshCircuitBreaker
@@ -25,7 +31,6 @@ type RefreshCoordinator struct {
 	config                 RefreshCoordinatorConfig
 	wg                     sync.WaitGroup
 	attemptsMutex          sync.RWMutex
-	refreshMutex           sync.RWMutex
 	cleanupTimerMu         sync.Mutex
 }
 
@@ -130,7 +135,7 @@ func NewRefreshCoordinator(config RefreshCoordinatorConfig, logger *Logger) *Ref
 	}
 
 	rc := &RefreshCoordinator{
-		inFlightRefreshes:      make(map[string]*refreshOperation),
+		// inFlightRefreshes is a sync.Map; zero value is ready to use.
 		sessionRefreshAttempts: make(map[string]*refreshAttemptTracker),
 		config:                 config,
 		metrics:                &RefreshMetrics{},
@@ -227,13 +232,28 @@ func (rc *RefreshCoordinator) getOrCreateOperation(
 	tokenHash string,
 	refreshToken string,
 ) (*refreshOperation, bool, error) {
-	rc.refreshMutex.Lock()
-	defer rc.refreshMutex.Unlock()
+	// Speculatively construct the operation we WOULD register if we win the
+	// race. Allocating here keeps the LoadOrStore call below atomic and
+	// avoids any global lock — under Yaegi the previous map+RWMutex design
+	// held the write lock long enough (tens of ms per call) that concurrent
+	// refreshes on the same coordinator serialized into a queue that grew
+	// without bound. See struct comment on inFlightRefreshes.
+	candidate := &refreshOperation{
+		refreshToken: refreshToken,
+		done:         make(chan struct{}),
+		startTime:    time.Now(),
+		waiterCount:  1,
+	}
 
-	// Check for existing operation while holding the lock
-	if existingOp, exists := rc.inFlightRefreshes[tokenHash]; exists {
+	if existing, loaded := rc.inFlightRefreshes.LoadOrStore(tokenHash, candidate); loaded {
+		existingOp, ok := existing.(*refreshOperation)
+		if !ok {
+			// Defensive: anything stored here is always *refreshOperation, but
+			// keep the typed assert so a programming error elsewhere doesn't
+			// surface as a confusing panic in an interpreter frame.
+			return nil, false, fmt.Errorf("inFlightRefreshes corrupt: unexpected type %T", existing)
+		}
 		if existingOp.refreshToken == refreshToken {
-			// Join existing operation
 			atomic.AddInt32(&existingOp.waiterCount, 1)
 			return existingOp, false, nil
 		}
@@ -241,41 +261,60 @@ func (rc *RefreshCoordinator) getOrCreateOperation(
 		return nil, false, fmt.Errorf("refresh token mismatch")
 	}
 
-	// No existing operation - check if we can create a new one
-	// All checks happen while holding the lock to prevent races
+	// We won the race and registered `candidate`. Apply gates now. If any
+	// gate fails we must remove our entry from the map and signal failure
+	// to any joiners that snuck in between LoadOrStore and now.
+	if err := rc.applyLeaderGates(sessionID); err != nil {
+		rc.failCandidate(tokenHash, candidate, err)
+		return nil, false, err
+	}
 
-	// Check and record refresh attempt for rate limiting
+	// Reserve concurrent slot via CAS — without the old global lock we can
+	// no longer rely on mutex-mediated check-then-increment. If we lose the
+	// CAS race we retry; if the limit has since been reached we back out.
+	for {
+		current := atomic.LoadInt32(&rc.metrics.currentInFlightRefreshes)
+		if int(current) >= rc.config.MaxConcurrentRefreshes {
+			err := fmt.Errorf("maximum concurrent refresh operations reached")
+			rc.failCandidate(tokenHash, candidate, err)
+			return nil, false, err
+		}
+		if atomic.CompareAndSwapInt32(&rc.metrics.currentInFlightRefreshes, current, current+1) {
+			break
+		}
+	}
+
+	return candidate, true, nil
+}
+
+// applyLeaderGates runs the rate-limit, cooldown, and memory-pressure checks
+// that previously ran under the global refreshMutex. Only the leader (the
+// goroutine that just registered the operation) runs them; joiners share the
+// leader's outcome via operation.done.
+func (rc *RefreshCoordinator) applyLeaderGates(sessionID string) error {
 	rc.recordRefreshAttempt(sessionID)
 	if rc.isInCooldown(sessionID) {
 		atomic.AddInt64(&rc.metrics.cooldownsTriggered, 1)
-		return nil, false, fmt.Errorf("refresh attempts exceeded for session, in cooldown period")
+		return fmt.Errorf("refresh attempts exceeded for session, in cooldown period")
 	}
-
-	// Check memory pressure
 	if rc.config.EnableMemoryPressureDetection && rc.isUnderMemoryPressure() {
 		atomic.AddInt64(&rc.metrics.memoryPressureEvents, 1)
-		return nil, false, fmt.Errorf("system under memory pressure, refresh denied")
+		return fmt.Errorf("system under memory pressure, refresh denied")
 	}
+	return nil
+}
 
-	// Check and reserve concurrent refresh slot atomically
-	current := atomic.LoadInt32(&rc.metrics.currentInFlightRefreshes)
-	if int(current) >= rc.config.MaxConcurrentRefreshes {
-		return nil, false, fmt.Errorf("maximum concurrent refresh operations reached")
-	}
-
-	// Reserve the slot - we're still holding the lock so this is safe
-	atomic.AddInt32(&rc.metrics.currentInFlightRefreshes, 1)
-
-	// Create and register new operation
-	operation := &refreshOperation{
-		refreshToken: refreshToken,
-		done:         make(chan struct{}),
-		startTime:    time.Now(),
-		waiterCount:  1,
-	}
-	rc.inFlightRefreshes[tokenHash] = operation
-
-	return operation, true, nil
+// failCandidate removes the leader's just-registered operation from the
+// in-flight map and signals the error to any joiners by recording the result
+// and closing the done channel. This keeps the (nil, false, err) return path
+// equivalent to the pre-sync.Map version: callers see the error directly,
+// joiners see it via operation.done.
+func (rc *RefreshCoordinator) failCandidate(tokenHash string, op *refreshOperation, err error) {
+	rc.inFlightRefreshes.Delete(tokenHash)
+	op.mutex.Lock()
+	op.result = &refreshResult{err: err}
+	op.mutex.Unlock()
+	close(op.done)
 }
 
 // executeRefreshAsync performs the actual refresh operation asynchronously
@@ -367,16 +406,11 @@ func (rc *RefreshCoordinator) scheduleDelayedCleanup(tokenHash string) {
 
 // performCleanup removes the operation from the in-flight map.
 // Idempotent: only decrements the in-flight counter if an entry was actually
-// removed. This guards against any future path accidentally calling cleanup
-// twice for the same tokenHash (which would corrupt the refresh budget).
+// removed. LoadAndDelete is atomic so any concurrent failCandidate or repeat
+// cleanup call will see exactly one removal — the budget cannot be corrupted
+// by double-decrement.
 func (rc *RefreshCoordinator) performCleanup(tokenHash string) {
-	rc.refreshMutex.Lock()
-	_, existed := rc.inFlightRefreshes[tokenHash]
-	if existed {
-		delete(rc.inFlightRefreshes, tokenHash)
-	}
-	rc.refreshMutex.Unlock()
-	if existed {
+	if _, existed := rc.inFlightRefreshes.LoadAndDelete(tokenHash); existed {
 		atomic.AddInt32(&rc.metrics.currentInFlightRefreshes, -1)
 	}
 }

telemetry.go

@@ -0,0 +1,142 @@
+package traefikoidc
+
+import (
+	"bytes"
+	"context"
+	"net/http"
+	"os"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+)
+
+// pluginVersion is bumped manually on each release. Keep in sync with the
+// most recent git tag (see `git tag --sort=-v:refname | head -1`).
+const pluginVersion = "1.0.11"
+
+const (
+	telemetryProject = "traefikoidc"
+	telemetryTimeout = 2 * time.Second
+)
+
+// telemetryEndpoint is intentionally a var rather than a const so the test
+// suite in this package can retarget it at an httptest server. Production
+// code never mutates it.
+var telemetryEndpoint = "https://oss.raczylo.com/v1/ping"
+
+// telemetryOnce guarantees a single anonymous "plugin loaded" ping per
+// process lifetime. Traefik can instantiate a middleware many times per
+// process (one per route using the plugin); the sync.Once gate keeps the
+// fire-and-forget call from amplifying into many pings.
+//
+// Reset in tests via `telemetryOnce = sync.Once{}`.
+var telemetryOnce sync.Once
+
+// telemetryInflight tracks any background goroutine started by sendTelemetry.
+// Tests Wait on it to drain in-flight goroutines before mutating package
+// state. Production code never calls Wait — the goroutine is fire-and-forget.
+var telemetryInflight sync.WaitGroup
+
+// sendTelemetry fires one anonymous usage ping in the background. It is
+// failproof by contract:
+//
+//   - never blocks the caller
+//   - never panics (the goroutine recovers internally)
+//   - never returns errors
+//   - silently dropped on invalid input, env-driven opt-out, or network failure
+//
+// Opt-out is honored via any of:
+//
+//   - DO_NOT_TRACK=1
+//   - OSS_TELEMETRY_DISABLED=1
+//   - TRAEFIKOIDC_DISABLE_TELEMETRY=1
+//
+// Yaegi note: this file deliberately avoids generics (atomic.Pointer[T]) and
+// range-over-int (Go 1.22) so it interprets under any reasonably recent
+// Traefik yaegi runtime.
+func sendTelemetry(version string) {
+	telemetryOnce.Do(func() {
+		if telemetryDisabledByEnv() {
+			return
+		}
+		if !validTelemetryVersion(version) {
+			return
+		}
+		telemetryInflight.Add(1)
+		go func() {
+			defer telemetryInflight.Done()
+			defer func() { _ = recover() }()
+			doTelemetryPost(version)
+		}()
+	})
+}
+
+func telemetryDisabledByEnv() bool {
+	keys := []string{
+		"DO_NOT_TRACK",
+		"OSS_TELEMETRY_DISABLED",
+		"TRAEFIKOIDC_DISABLE_TELEMETRY",
+	}
+	for _, k := range keys {
+		v := strings.ToLower(strings.TrimSpace(os.Getenv(k)))
+		if v == "1" || v == "true" || v == "yes" || v == "on" {
+			return true
+		}
+	}
+	return false
+}
+
+// validTelemetryVersion mirrors the server-side regex ^[A-Za-z0-9.+_-]{1,32}$
+// using a byte loop. No allocation, no regexp dependency.
+//
+// Yaegi note: written as an `||` chain rather than `switch{case A,B,C:}` —
+// some yaegi releases mis-evaluate comma-separated case expressions in
+// switch-true blocks, returning false for all inputs.
+func validTelemetryVersion(v string) bool {
+	if len(v) == 0 || len(v) > 32 {
+		return false
+	}
+	for i := 0; i < len(v); i++ {
+		c := v[i]
+		ok := (c >= 'A' && c <= 'Z') ||
+			(c >= 'a' && c <= 'z') ||
+			(c >= '0' && c <= '9') ||
+			c == '.' || c == '+' || c == '_' || c == '-'
+		if !ok {
+			return false
+		}
+	}
+	return true
+}
+
+// doTelemetryPost builds the JSON body manually. The project name is a
+// constant and the version is pre-validated against an ASCII-only allowlist,
+// so direct concatenation needs no JSON escaping.
+func doTelemetryPost(version string) {
+	body := make([]byte, 0, 96)
+	body = append(body, `{"project":"`...)
+	body = append(body, telemetryProject...)
+	body = append(body, `","version":"`...)
+	body = append(body, version...)
+	body = append(body, `","ts":`...)
+	body = strconv.AppendInt(body, time.Now().Unix(), 10)
+	body = append(body, '}')
+
+	ctx, cancel := context.WithTimeout(context.Background(), telemetryTimeout)
+	defer cancel()
+
+	url := telemetryEndpoint
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body))
+	if err != nil {
+		return
+	}
+	req.Header.Set("Content-Type", "application/json")
+
+	client := &http.Client{Timeout: telemetryTimeout}
+	resp, err := client.Do(req)
+	if err != nil {
+		return
+	}
+	_ = resp.Body.Close()
+}

telemetry_test.go

@@ -0,0 +1,167 @@
+package traefikoidc
+
+import (
+	"encoding/json"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+// resetTelemetryState restores package-level mutable state so tests do not
+// contaminate one another. The cleanup waits for any in-flight ping goroutine
+// to finish before restoring telemetryEndpoint — without that drain step the
+// goroutine and the cleanup would race on the var.
+func resetTelemetryState(t *testing.T) {
+	t.Helper()
+	telemetryOnce = sync.Once{}
+	prev := telemetryEndpoint
+	t.Cleanup(func() {
+		telemetryInflight.Wait()
+		telemetryEndpoint = prev
+		telemetryOnce = sync.Once{}
+	})
+}
+
+func newTelemetryServer(t *testing.T, status int) (hits *int32, lastBody func() string) {
+	t.Helper()
+	var counter int32
+	var mu sync.Mutex
+	var body string
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		atomic.AddInt32(&counter, 1)
+		b, _ := io.ReadAll(r.Body)
+		_ = r.Body.Close()
+		mu.Lock()
+		body = string(b)
+		mu.Unlock()
+		w.WriteHeader(status)
+	}))
+	telemetryEndpoint = srv.URL
+	t.Cleanup(srv.Close)
+	return &counter, func() string {
+		mu.Lock()
+		defer mu.Unlock()
+		return body
+	}
+}
+
+func TestValidTelemetryVersion(t *testing.T) {
+	good := []string{"1.2.3", "1.4.0-beta1", "2.0", "v1.0.0", "1.0.0+meta", "dev"}
+	for _, v := range good {
+		if !validTelemetryVersion(v) {
+			t.Errorf("validTelemetryVersion(%q) = false, want true", v)
+		}
+	}
+	bad := []string{"", "has space", "semi;colon", strings.Repeat("1", 33)}
+	for _, v := range bad {
+		if validTelemetryVersion(v) {
+			t.Errorf("validTelemetryVersion(%q) = true, want false", v)
+		}
+	}
+}
+
+func TestTelemetryDisabledByEnv(t *testing.T) {
+	for _, k := range []string{"DO_NOT_TRACK", "OSS_TELEMETRY_DISABLED", "TRAEFIKOIDC_DISABLE_TELEMETRY"} {
+		t.Run(k, func(t *testing.T) {
+			t.Setenv(k, "1")
+			if !telemetryDisabledByEnv() {
+				t.Fatalf("%s=1 should disable", k)
+			}
+		})
+	}
+	t.Run("falsy_values_do_not_disable", func(t *testing.T) {
+		t.Setenv("DO_NOT_TRACK", "0")
+		t.Setenv("OSS_TELEMETRY_DISABLED", "false")
+		t.Setenv("TRAEFIKOIDC_DISABLE_TELEMETRY", "no")
+		if telemetryDisabledByEnv() {
+			t.Fatal("falsy env values should not disable")
+		}
+	})
+}
+
+func TestSendTelemetry_FiresOnceAcrossManyCalls(t *testing.T) {
+	resetTelemetryState(t)
+	hits, lastBody := newTelemetryServer(t, http.StatusNoContent)
+
+	for i := 0; i < 50; i++ {
+		sendTelemetry("1.2.3")
+	}
+	telemetryInflight.Wait()
+
+	if got := atomic.LoadInt32(hits); got != 1 {
+		t.Fatalf("expected exactly 1 hit, got %d", got)
+	}
+
+	var payload struct {
+		Project string `json:"project"`
+		Version string `json:"version"`
+		Ts      int64  `json:"ts"`
+	}
+	if err := json.Unmarshal([]byte(lastBody()), &payload); err != nil {
+		t.Fatalf("server received non-JSON body: %q (err: %v)", lastBody(), err)
+	}
+	if payload.Project != "traefikoidc" || payload.Version != "1.2.3" || payload.Ts <= 0 {
+		t.Fatalf("unexpected payload: %+v", payload)
+	}
+}
+
+func TestSendTelemetry_RespectsDisableEnv(t *testing.T) {
+	resetTelemetryState(t)
+	hits, _ := newTelemetryServer(t, http.StatusNoContent)
+	t.Setenv("DO_NOT_TRACK", "1")
+
+	sendTelemetry("1.2.3")
+	telemetryInflight.Wait()
+
+	if got := atomic.LoadInt32(hits); got != 0 {
+		t.Fatalf("DO_NOT_TRACK should suppress; got %d hits", got)
+	}
+}
+
+func TestSendTelemetry_DropsInvalidVersion(t *testing.T) {
+	resetTelemetryState(t)
+	hits, _ := newTelemetryServer(t, http.StatusNoContent)
+
+	sendTelemetry("has space")
+	telemetryInflight.Wait()
+
+	if got := atomic.LoadInt32(hits); got != 0 {
+		t.Fatalf("invalid version should suppress; got %d hits", got)
+	}
+}
+
+func TestSendTelemetry_DoesNotBlock(t *testing.T) {
+	resetTelemetryState(t)
+	// Hanging server proves the caller is never blocked. The 2s context
+	// timeout in doTelemetryPost ensures the goroutine eventually exits;
+	// resetTelemetryState's cleanup waits for that drain before restoring
+	// telemetryEndpoint so there is no race with this test's mutation.
+	hung := httptest.NewServer(http.HandlerFunc(func(_ http.ResponseWriter, _ *http.Request) {
+		time.Sleep(5 * time.Second)
+	}))
+	t.Cleanup(hung.Close)
+	telemetryEndpoint = hung.URL
+
+	start := time.Now()
+	sendTelemetry("1.2.3")
+	if elapsed := time.Since(start); elapsed > 50*time.Millisecond {
+		t.Fatalf("sendTelemetry blocked for %v, expected near-instant return", elapsed)
+	}
+}
+
+func TestSendTelemetry_SurvivesServerError(t *testing.T) {
+	resetTelemetryState(t)
+	hits, _ := newTelemetryServer(t, http.StatusInternalServerError)
+
+	sendTelemetry("1.2.3")
+	telemetryInflight.Wait()
+
+	if got := atomic.LoadInt32(hits); got != 1 {
+		t.Fatalf("request should still reach server even on 500; got %d hits", got)
+	}
+}