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

Five failed asks to waf within five seconds open the breaker. An ask
fails when it gets no answer in time, slow answers included, or an
answer rproxy cannot pass on. Every usable answer clears the failures
and closes the breaker. While open, a request takes
SWW_RPROXY_WAF_BACKEND_NO_ANSWER_DISPOSITION without asking waf, except
one ask a second. 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,139 @@ 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 TestAnswersThatCannotBePassedOnOpenTheBreaker(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, out := startRProxy(t, standIn.URL, app.URL, map[string]string{
+		wafBackendTimeout: "60000",
+	})
+
+	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, "/"))
+	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 TestAPathUnderOwnPrefixNeverReachesTheApp(t *testing.T) {
 	t.Parallel()