🐛 keep the stream and metrics tickers on their own channels (#399)

Traefik runs the plugin under yaegi, which shares the case list of a
select statement between every goroutine that runs it. Both ticker
loops ran the same select, so when the tickers fired together one loop
could end up waiting on the other's channel, dropping stream ticks for
up to two metrics intervals.

Range over the ticker channel instead, which yaegi evaluates per
execution. The stop channel was never used.

Fixes #377

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

bouncer.go

@@ -70,8 +70,8 @@ var (
 	isCrowdsecStreamStartup = true
 	isCrowdsecStreamHealthy = true
 	updateFailure           int64
-	streamTicker            chan bool
+	streamTicker            *time.Ticker
-	metricsTicker           chan bool
+	metricsTicker           *time.Ticker
 	lastMetricsPush         time.Time
 	blockedRequests         int64
 )
@@ -324,7 +324,7 @@ func New(_ context.Context, next http.Handler, config *configuration.Config, nam
 		} else {
 			go handleStreamTicker(bouncer)
 		}
-		streamTicker = startTicker("stream", config.UpdateIntervalSeconds, log, func() {
+		streamTicker = startTicker(config.UpdateIntervalSeconds, func() {
 			handleStreamTicker(bouncer)
 		})
 	}
@@ -333,7 +333,7 @@ func New(_ context.Context, next http.Handler, config *configuration.Config, nam
 	if metricsTicker == nil && config.MetricsUpdateIntervalSeconds > 0 {
 		lastMetricsPush = time.Now() // Initialize lastMetricsPush when starting the metrics ticker
 		go handleMetricsTicker(bouncer)
-		metricsTicker = startTicker("metrics", config.MetricsUpdateIntervalSeconds, log, func() {
+		metricsTicker = startTicker(config.MetricsUpdateIntervalSeconds, func() {
 			handleMetricsTicker(bouncer)
 		})
 	}
@@ -564,21 +564,16 @@ func handleMetricsTicker(bouncer *Bouncer) {
 	}
 }
 
-func startTicker(name string, updateInterval int64, log *slog.Logger, work func()) chan bool {
+func startTicker(updateInterval int64, work func()) *time.Ticker {
 	ticker := time.NewTicker(time.Duration(updateInterval) * time.Second)
-	stop := make(chan bool, 1)
+	go runTicker(ticker.C, work)
-	go func() {
+	return ticker
-		defer log.Debug(name + "_ticker:stopped")
+}
-		for {
+
-			select {
+func runTicker(ticks <-chan time.Time, work func()) {
-			case <-ticker.C:
+	for range ticks {
-				go work()
+		go work()
-			case <-stop:
+	}
-				return
-			}
-		}
-	}()
-	return stop
 }
 
 // We are now in none or live mode.

ticker_test.go

@@ -0,0 +1,37 @@
+package crowdsec_bouncer_traefik_plugin //nolint:revive,stylecheck
+
+import (
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func sendTicks(ticks chan time.Time, count int) {
+	for i := 0; i < count; i++ { //nolint:intrange
+		ticks <- time.Time{}
+	}
+}
+
+func Test_runTicker_keepsEachTickerOnItsOwnChannel(t *testing.T) {
+	const count = 20000
+	var streamRuns, metricsRuns int64
+	var runs sync.WaitGroup
+	runs.Add(2 * count)
+	streamTicks := make(chan time.Time)
+	metricsTicks := make(chan time.Time)
+	go runTicker(streamTicks, func() {
+		atomic.AddInt64(&streamRuns, 1)
+		runs.Done()
+	})
+	go runTicker(metricsTicks, func() {
+		atomic.AddInt64(&metricsRuns, 1)
+		runs.Done()
+	})
+	go sendTicks(streamTicks, count)
+	go sendTicks(metricsTicks, count)
+	runs.Wait()
+	if streamRuns != count || metricsRuns != count {
+		t.Errorf("runTicker ran stream %d and metrics %d times, want %d each", streamRuns, metricsRuns, count)
+	}
+}