✨️ Do not block middleware startup due to stream intialization (#281)

* 🍱 not block on stream startup

* 🍱 fix

* 🍱 fix lint

* 🍱 fix lint

* 🍱 fix lint

* 🍱 update readme

* 📚 clarify StreamStartupBlock warning in README

Make the fail-open implication of StreamStartupBlock=false explicit:
banned IPs are allowed through until the first stream sync completes.

* 🐛 clear isCrowdsecStreamStartup on alreadyUpdated path

When cacheTimeoutKey is already set (another instance/process has
populated the stream cache within the update window), the early return
in handleStreamCache used to leave isCrowdsecStreamStartup at true,
causing this instance to keep sending startup=true on subsequent ticks.
Flip the flag in the early-return path so startup is correctly tracked
across multi-instance deployments sharing a cache.

---------

Co-authored-by: mhx <mathieu@hanotaux.fr>

README.md

@@ -465,6 +465,12 @@ make run
   - int64
   - default: 0
   - Used only in `stream` and `alone` mode, the maximum number of time we can not reach Crowdsec before blocking traffic (set -1 to never block)
+- StreamStartupBlock
+  - bool
+  - default: true
+  - Used only in `stream` and `alone` mode, controls whether the initial stream update runs synchronously or asynchronously during plugin initialization
+  - When `true`, plugin initialization waits for Crowdsec to be ready before serving traffic.
+  - **Warning**: When `false`, all requests bypass remediation until the first stream sync completes — banned IPs will be allowed through during this window. Only disable when startup availability is more important than blocking at startup.
 - DefaultDecisionSeconds
   - int64
   - default: 60
@@ -598,6 +604,7 @@ http:
           LogFilePath: ""
           updateIntervalSeconds: 60
           updateMaxFailure: 0
+          streamStartupBlock: true
           defaultDecisionSeconds: 60
           remediationStatusCode: 403
           httpTimeoutSeconds: 10

bouncer.go

@@ -64,7 +64,7 @@ const (
 
 //nolint:gochecknoglobals
 var (
-	isStartup               = true
+	isCrowdsecStreamStartup = true
 	isCrowdsecStreamHealthy = true
 	updateFailure           int64
 	streamTicker            chan bool
@@ -123,7 +123,7 @@ type Bouncer struct {
 
 // New creates the crowdsec bouncer plugin.
 //
-//nolint:gocyclo
+//nolint:nestif,gocyclo,gocognit
 func New(_ context.Context, next http.Handler, config *configuration.Config, name string) (http.Handler, error) {
 	config.LogLevel = strings.ToUpper(config.LogLevel)
 	log := logger.NewWithFormat(config.LogLevel, config.LogFilePath, config.LogFormat)
@@ -291,8 +291,11 @@ func New(_ context.Context, next http.Handler, config *configuration.Config, nam
 				return nil, err
 			}
 		}
-		handleStreamTicker(bouncer)
-		isStartup = false
+		if config.StreamStartupBlock {
+			handleStreamTicker(bouncer)
+		} else {
+			go handleStreamTicker(bouncer)
+		}
 		streamTicker = startTicker("stream", config.UpdateIntervalSeconds, log, func() {
 			handleStreamTicker(bouncer)
 		})
@@ -301,7 +304,7 @@ func New(_ context.Context, next http.Handler, config *configuration.Config, nam
 	// Start metrics ticker if not already running
 	if metricsTicker == nil && config.MetricsUpdateIntervalSeconds > 0 {
 		lastMetricsPush = time.Now() // Initialize lastMetricsPush when starting the metrics ticker
-		handleMetricsTicker(bouncer)
+		go handleMetricsTicker(bouncer)
 		metricsTicker = startTicker("metrics", config.MetricsUpdateIntervalSeconds, log, func() {
 			handleMetricsTicker(bouncer)
 		})
@@ -624,6 +627,7 @@ func handleStreamCache(bouncer *Bouncer) error {
 	_, err := bouncer.cacheClient.Get(cacheTimeoutKey)
 	if err == nil {
 		bouncer.log.Debug("handleStreamCache:alreadyUpdated")
+		isCrowdsecStreamStartup = false
 		return nil
 	}
 	if err.Error() != cache.CacheMiss {
@@ -634,7 +638,7 @@ func handleStreamCache(bouncer *Bouncer) error {
 		Scheme:   bouncer.crowdsecScheme,
 		Host:     bouncer.crowdsecHost,
 		Path:     bouncer.crowdsecPath + bouncer.crowdsecStreamRoute,
-		RawQuery: fmt.Sprintf("startup=%t", !isCrowdsecStreamHealthy || isStartup),
+		RawQuery: fmt.Sprintf("startup=%t", !isCrowdsecStreamHealthy || isCrowdsecStreamStartup),
 	}
 	body, err := crowdsecQuery(bouncer, streamRouteURL.String(), nil)
 	if err != nil {
@@ -664,6 +668,7 @@ func handleStreamCache(bouncer *Bouncer) error {
 		bouncer.cacheClient.Delete(decision.Value)
 	}
 	bouncer.log.Debug("handleStreamCache:updated")
+	isCrowdsecStreamStartup = false
 	return nil
 }
 

pkg/configuration/configuration.go

@@ -84,6 +84,7 @@ type Config struct {
 	UpdateIntervalSeconds                      int64    `json:"updateIntervalSeconds,omitempty"`
 	MetricsUpdateIntervalSeconds               int64    `json:"metricsUpdateIntervalSeconds,omitempty"`
 	UpdateMaxFailure                           int64    `json:"updateMaxFailure,omitempty"`
+	StreamStartupBlock                         bool     `json:"streamStartupBlock,omitempty"`
 	DefaultDecisionSeconds                     int64    `json:"defaultDecisionSeconds,omitempty"`
 	RemediationStatusCode                      int      `json:"remediationStatusCode,omitempty"`
 	HTTPTimeoutSeconds                         int64    `json:"httpTimeoutSeconds,omitempty"`
@@ -146,6 +147,7 @@ func New() *Config {
 		UpdateIntervalSeconds:           60,
 		MetricsUpdateIntervalSeconds:    600,
 		UpdateMaxFailure:                0,
+		StreamStartupBlock:              true,
 		DefaultDecisionSeconds:          60,
 		RemediationStatusCode:           http.StatusForbidden,
 		HTTPTimeoutSeconds:              10,