rproxy: circuit breaker for asks to waf (closes #392)

Five asks that get no answer in time within five seconds open the
breaker. While open, a request takes
SWW_RPROXY_WAF_BACKEND_NO_ANSWER_DISPOSITION without asking waf, except
one ask a second; the first answer closes it. Slow answers count as
failures. SPEC.md describes it.

The sww test that stops and restarts waf now waits for the breaker to
close before it expects the ban again.

Model: opus-5-5

internal/rproxy/rproxy_test.go

@@ -15,6 +15,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -1395,6 +1396,134 @@ func TestSendsNoReportWithoutAnAnswer(t *testing.T) {
 	standIn.wantNoReport(t)
 }
 
+func TestFiveFailedAsksKeepTheNextRequestFromAskingWAF(t *testing.T) {
+	t.Parallel()
+
+	for name, failure := range map[string]http.HandlerFunc{
+		"no answer at all": func(http.ResponseWriter, *http.Request) {
+			panic(http.ErrAbortHandler)
+		},
+		"a slow answer": func(_ http.ResponseWriter, r *http.Request) {
+			<-r.Context().Done()
+		},
+	} {
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+
+			var answering atomic.Bool
+
+			standIn := startWAF(
+				t,
+				func(w http.ResponseWriter, r *http.Request) {
+					if answering.Load() {
+						allow(w, r)
+
+						return
+					}
+
+					failure(w, r)
+				},
+			)
+			app := startApp(t, greet)
+			addr, out := startRProxy(
+				t,
+				standIn.URL,
+				app.URL,
+				map[string]string{
+					wafBackendTimeout: "200",
+				},
+			)
+
+			for range 5 {
+				wantGreeting(t, get(t, addr, "/"))
+				standIn.question(t)
+			}
+
+			answering.Store(true)
+			wantGreeting(t, get(t, addr, "/"))
+			out.waitFor(t, "the circuit breaker is open")
+
+			select {
+			case <-standIn.questions:
+				t.Error("waf was asked, want it not to be")
+			case <-time.After(10 * pollInterval):
+			}
+		})
+	}
+}
+
+func TestVisitorsThatGoAwayWhileWAFIsAskedLeaveTheBreakerClosed(
+	t *testing.T,
+) {
+	t.Parallel()
+
+	var answering atomic.Bool
+
+	standIn := startWAF(t, func(w http.ResponseWriter, r *http.Request) {
+		if answering.Load() {
+			allow(w, r)
+
+			return
+		}
+
+		<-r.Context().Done()
+	})
+	app := startApp(t, greet)
+	addr, out := startRProxy(t, standIn.URL, app.URL, map[string]string{
+		wafBackendTimeout: "60000",
+	})
+
+	// Five of these would open the breaker if they counted as failures.
+	for range 5 {
+		conn := dial(t, addr)
+		send(t, conn, "GET / HTTP/1.1\r\nHost: site.example\r\n\r\n")
+		id := standIn.question(t).header.Get(requestIDHeader)
+
+		_ = conn.Close()
+
+		// rproxy writes the request's log line once it is done with the ask.
+		out.waitFor(t, id)
+	}
+
+	answering.Store(true)
+	wantGreeting(t, get(t, addr, "/"))
+	standIn.question(t)
+}
+
+func TestAnswersThatCannotBePassedOnLeaveTheBreakerClosed(t *testing.T) {
+	t.Parallel()
+
+	var answering atomic.Bool
+
+	tooLarge := overOneMiB(dispositionDeny)
+	standIn := startWAF(t, func(w http.ResponseWriter, r *http.Request) {
+		if answering.Load() {
+			allow(w, r)
+
+			return
+		}
+
+		tooLarge(w, r)
+	})
+	app := startApp(t, greet)
+	addr, _ := startRProxy(t, standIn.URL, app.URL, map[string]string{
+		wafBackendTimeout: "60000",
+	})
+
+	// Five of these would open the breaker if they counted as failures.
+	for range 5 {
+		if got := get(t, addr, "/"); got.status != http.StatusBadGateway {
+			t.Errorf("answered %d, want 502", got.status)
+		}
+
+		standIn.question(t)
+	}
+
+	answering.Store(true)
+	wantGreeting(t, get(t, addr, "/"))
+	standIn.question(t)
+}
+
 func TestAPathUnderOwnPrefixNeverReachesTheApp(t *testing.T) {
 	t.Parallel()