s3api: add optional request interceptor to circuit breaker (#9994)

* s3api: add optional request interceptor to circuit breaker

Add an optional Interceptor func(next http.HandlerFunc, action string) http.HandlerFunc
field on CircuitBreaker, applied at the very top of Limit() -- before upload
concurrency limiting and before the 'if !cb.Enabled' early return -- so it runs
for every route regardless of breaker state.

It is nil by default (no behavior change). An interceptor may reject a request
(write its own response and skip next) or wrap next to observe/shape it. This
provides a single per-request extension point at the existing per-route
chokepoint (cb.Limit already wraps nearly every S3 route), useful for tracing,
auditing, or request rate limiting without touching the route table.

* s3api: evaluate circuit breaker interceptor per request

Move the Interceptor nil-check into the returned handler so it is consulted
per request instead of captured at registration time. This:
- lets the interceptor be installed after registerRouter runs (handlers are
  built during construction, before request-time dependencies exist), and
- avoids dereferencing a nil CircuitBreaker at registration time, which
  panicked tests that register routes on a server with a nil cb
  (e.g. TestRouting_STSWithQueryParams).

Adds a regression test for installing the interceptor after Limit() returns.

weed/s3api/s3api_circuit_breaker.go

@@ -25,6 +25,14 @@ type CircuitBreaker struct {
 	counters    map[string]*int64
 	limitations map[string]int64
 	s3a         *S3ApiServer
+
+	// Interceptor, if set, wraps the per-route handler ahead of ALL
+	// circuit-breaker logic (upload limiting and the breaker checks) and runs
+	// even when the breaker is disabled. It is a general per-request
+	// interceptor seam: an implementation may reject the request (write its own
+	// response and not call next) or wrap next to observe/shape it. Nil by
+	// default, so it is a no-op unless explicitly set.
+	Interceptor func(next http.HandlerFunc, action string) http.HandlerFunc
 }
 
 func NewCircuitBreaker(option *S3ApiServerOption) *CircuitBreaker {
@@ -92,7 +100,7 @@ func (cb *CircuitBreaker) loadCircuitBreakerConfig(cfg *s3_pb.S3CircuitBreakerCo
 }
 
 func (cb *CircuitBreaker) Limit(f func(w http.ResponseWriter, r *http.Request), action string) (http.HandlerFunc, Action) {
-	return func(w http.ResponseWriter, r *http.Request) {
+	inner := func(w http.ResponseWriter, r *http.Request) {
 		// Apply upload limiting for write actions if configured
 		if cb.s3a != nil && (action == s3_constants.ACTION_WRITE) &&
 			(cb.s3a.option.ConcurrentUploadLimit != 0 || cb.s3a.option.ConcurrentFileUploadLimit != 0) {
@@ -161,6 +169,20 @@ func (cb *CircuitBreaker) Limit(f func(w http.ResponseWriter, r *http.Request),
 			return
 		}
 		s3err.WriteErrorResponse(w, r, errCode)
+	}
+
+	// The interceptor is consulted per request rather than captured here, so it
+	// can be installed after the routes are registered (e.g. once the server and
+	// its dependencies are constructed) and so a nil CircuitBreaker is never
+	// dereferenced at registration time. When unset this is just a nil check.
+	// It runs outermost: before upload limiting and the breaker checks, and
+	// regardless of cb.Enabled.
+	return func(w http.ResponseWriter, r *http.Request) {
+		if cb.Interceptor != nil {
+			cb.Interceptor(inner, action)(w, r)
+			return
+		}
+		inner(w, r)
 	}, Action(action)
 }
 

weed/s3api/s3api_circuit_breaker_test.go

@@ -2,6 +2,7 @@ package s3api
 
 import (
 	"net/http"
+	"net/http/httptest"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -106,3 +107,99 @@ func doLimit(circuitBreaker *CircuitBreaker, routineCount int, r *http.Request,
 	}
 	return successCounter
 }
+
+// TestLimitInterceptor verifies the optional request interceptor: it is a no-op
+// when nil, runs ahead of the (disabled) breaker logic, and can either reject a
+// request or pass it through to the wrapped handler.
+func TestLimitInterceptor(t *testing.T) {
+	readAction := s3_constants.ACTION_READ
+	newCB := func() *CircuitBreaker {
+		// Enabled defaults to false and s3a is nil, so without an interceptor
+		// Limit's handler falls straight through to the wrapped handler.
+		return &CircuitBreaker{counters: make(map[string]*int64), limitations: make(map[string]int64)}
+	}
+
+	// 1. nil interceptor must not change behavior: the handler still runs.
+	t.Run("nil interceptor is a no-op", func(t *testing.T) {
+		cb := newCB()
+		called := false
+		h, _ := cb.Limit(func(w http.ResponseWriter, r *http.Request) { called = true }, readAction)
+		h(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, "/bucket/object", nil))
+		if !called {
+			t.Fatal("handler should run when no interceptor is set")
+		}
+	})
+
+	// 2. a rejecting interceptor runs first and prevents the handler, even
+	//    though the breaker itself is disabled.
+	t.Run("interceptor can reject", func(t *testing.T) {
+		cb := newCB()
+		var order []string
+		cb.Interceptor = func(next http.HandlerFunc, action string) http.HandlerFunc {
+			return func(w http.ResponseWriter, r *http.Request) {
+				order = append(order, "interceptor")
+				s3err.WriteErrorResponse(w, r, s3err.ErrTooManyRequest) // reject; do not call next
+			}
+		}
+		handlerRan := false
+		h, _ := cb.Limit(func(w http.ResponseWriter, r *http.Request) {
+			handlerRan = true
+			order = append(order, "handler")
+		}, readAction)
+		rec := httptest.NewRecorder()
+		h(rec, httptest.NewRequest(http.MethodGet, "/bucket/object", nil))
+		if handlerRan {
+			t.Fatal("a rejecting interceptor must prevent the handler from running")
+		}
+		if rec.Code != http.StatusServiceUnavailable {
+			t.Fatalf("expected 503 from ErrTooManyRequest, got %d", rec.Code)
+		}
+		if len(order) != 1 || order[0] != "interceptor" {
+			t.Fatalf("interceptor should run alone, got %v", order)
+		}
+	})
+
+	// 3. a pass-through interceptor runs before the handler and sees the action.
+	t.Run("interceptor can pass through", func(t *testing.T) {
+		cb := newCB()
+		var order []string
+		var seenAction string
+		cb.Interceptor = func(next http.HandlerFunc, action string) http.HandlerFunc {
+			return func(w http.ResponseWriter, r *http.Request) {
+				seenAction = action
+				order = append(order, "interceptor")
+				next(w, r)
+			}
+		}
+		h, _ := cb.Limit(func(w http.ResponseWriter, r *http.Request) {
+			order = append(order, "handler")
+		}, readAction)
+		h(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, "/bucket/object", nil))
+		if seenAction != readAction {
+			t.Fatalf("interceptor should receive the route action, got %q", seenAction)
+		}
+		if len(order) != 2 || order[0] != "interceptor" || order[1] != "handler" {
+			t.Fatalf("interceptor must run before the handler, got %v", order)
+		}
+	})
+
+	// 4. installed AFTER Limit() returns: still takes effect, because the
+	//    interceptor is consulted per request rather than captured at
+	//    registration time (the handlers are built during router registration,
+	//    before dependencies that need the running server exist).
+	t.Run("interceptor installed after registration takes effect", func(t *testing.T) {
+		cb := newCB()
+		h, _ := cb.Limit(func(w http.ResponseWriter, r *http.Request) {}, readAction) // built while nil
+		ran := false
+		cb.Interceptor = func(next http.HandlerFunc, action string) http.HandlerFunc {
+			return func(w http.ResponseWriter, r *http.Request) {
+				ran = true
+				next(w, r)
+			}
+		}
+		h(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, "/bucket/object", nil))
+		if !ran {
+			t.Fatal("interceptor set after Limit() must still run (request-time evaluation)")
+		}
+	})
+}